diff --git a/alembic/versions/0118_rule_moment_judgments.py b/alembic/versions/0118_rule_moment_judgments.py
new file mode 100644
index 00000000..3884ea0e
--- /dev/null
+++ b/alembic/versions/0118_rule_moment_judgments.py
@@ -0,0 +1,61 @@
+"""rule_moment_judgments — whether a rule belongs on a moment, and who said so
+(milestone 458 step 7, #4925)
+
+Revision ID: 0118
+Revises: 0117
+Create Date: 2026-10-05
+
+One row per (rule, moment) with a state: `suggested` while a proposal waits
+for a judgment, `confirmed` or `rejected` once one is made. The mount itself
+stays in rule_moments; this records the proposals that are not mounts yet and
+the rejections that must stop them being proposed again. Moment '' is "no
+moment fits this rule". CASCADE on the rule — a judgment about a rule that no
+longer exists says nothing. No backfill: the mounts made before this table
+were judgments too, but recording them now would invent who made them.
+"""
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+from alembic import op
+
+revision = "0118"
+down_revision = "0117"
+branch_labels = None
+depends_on = None
+
+# One place each, so the CHECKs and the model's JUDGMENT_STATES / SOURCES
+# cannot drift (rule 36: a new value later means DROP + ADD CONSTRAINT in the
+# same migration).
+_STATES = ("suggested", "confirmed", "rejected")
+_SOURCES = ("pass", "signal", "edit")
+
+
+def upgrade() -> None:
+    op.create_table(
+        "rule_moment_judgments",
+        sa.Column("id", sa.BigInteger(), primary_key=True),
+        sa.Column("rule_id", sa.BigInteger(), sa.ForeignKey("rules.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("moment", sa.Text(), nullable=False),
+        sa.Column("state", sa.Text(), nullable=False, server_default="suggested"),
+        sa.Column("source", sa.Text(), nullable=False, server_default="pass"),
+        sa.Column("note", sa.Text(), nullable=True),
+        sa.Column("evidence", postgresql.JSONB(), nullable=True),
+        sa.Column("judged_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")),
+        sa.UniqueConstraint("rule_id", "moment", name="uq_rule_moment_judgments_pair"),
+    )
+    op.create_check_constraint(
+        "ck_rule_moment_judgments_state", "rule_moment_judgments",
+        "state IN (" + ", ".join(f"'{s}'" for s in _STATES) + ")",
+    )
+    op.create_check_constraint(
+        "ck_rule_moment_judgments_source", "rule_moment_judgments",
+        "source IN (" + ", ".join(f"'{s}'" for s in _SOURCES) + ")",
+    )
+    op.create_index("ix_rule_moment_judgments_rule_id", "rule_moment_judgments", ["rule_id"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_rule_moment_judgments_rule_id", table_name="rule_moment_judgments")
+    op.drop_constraint("ck_rule_moment_judgments_source", "rule_moment_judgments", type_="check")
+    op.drop_constraint("ck_rule_moment_judgments_state", "rule_moment_judgments", type_="check")
+    op.drop_table("rule_moment_judgments")
diff --git a/frontend/src/api/moments.ts b/frontend/src/api/moments.ts
index 9a8f8581..70dd0b07 100644
--- a/frontend/src/api/moments.ts
+++ b/frontend/src/api/moments.ts
@@ -99,3 +99,59 @@ export function unmapAction(change: MappingChange): Promise<void> {
   });
   return apiDelete(`/api/retrieval/moments/mappings?${q.toString()}`);
 }
+
+/**
+ * Proposals that a rule belongs on a moment (milestone 458 step 7) — from a
+ * pass that read the rule, or from the rule being opened just after the
+ * moment fired. `GET /api/retrieval/moments/proposals`, the payload the
+ * `rule_moment_proposals` MCP tool returns.
+ */
+export type ProposalSource = "pass" | "signal" | "edit";
+
+export interface MomentProposal {
+  moment: string;
+  source: ProposalSource;
+  why: string;
+  evidence: { situations: number; projects: number; co_surfaced: number };
+  created_at: string | null;
+}
+
+export interface RuleProposals {
+  id: number;
+  title: string;
+  kind: string;
+  statement: string;
+  when_to_apply: string;
+  home: "global" | "project";
+  /** What the rule is mounted on now. */
+  mounted: string[];
+  proposals: MomentProposal[];
+}
+
+export interface ProposalsPayload {
+  rules: RuleProposals[];
+  total: number;
+}
+
+export type Verdict = "confirm" | "reject";
+
+export interface MomentJudgment {
+  rule_id: number;
+  moment: string;
+  verdict: Verdict;
+  note?: string;
+}
+
+export interface JudgeResult {
+  judged: { rule_id: number; moment: string; state: string }[];
+  refused: { rule_id: number; moment?: string; error: string }[];
+}
+
+export function getProposals(ruleId?: number): Promise<ProposalsPayload> {
+  return apiGet(`/api/retrieval/moments/proposals${ruleId ? `?rule_id=${ruleId}` : ""}`);
+}
+
+/** A confirm MOUNTS the rule on the moment; a reject stops it being proposed again. */
+export function judgeProposals(judgments: MomentJudgment[]): Promise<JudgeResult> {
+  return apiPost("/api/retrieval/moments/proposals/judge", { judgments });
+}
diff --git a/frontend/src/components/MomentProposals.vue b/frontend/src/components/MomentProposals.vue
new file mode 100644
index 00000000..2250bf29
--- /dev/null
+++ b/frontend/src/components/MomentProposals.vue
@@ -0,0 +1,138 @@
+<script setup lang="ts">
+import { computed, onMounted, ref } from "vue";
+import { getProposals, judgeProposals } from "@/api/moments";
+import type { MomentProposal, ProposalsPayload, RuleProposals, Verdict } from "@/api/moments";
+import { apiErrorMessage } from "@/api/client";
+import { useMomentsStore } from "@/stores/moments";
+import { useToastStore } from "@/stores/toast";
+
+/**
+ * Proposals that a rule belongs on a moment, waiting on a person (milestone
+ * 458 step 7). They come from a pass that read the rule, or from the rule
+ * being opened just after the moment fired in several sessions. Nothing here
+ * is mounted until "Mount" is pressed — the same judgment `judge_rule_moments`
+ * makes in a session, through the same service.
+ */
+const store = useMomentsStore();
+const toast = useToastStore();
+
+const data = ref<ProposalsPayload | null>(null);
+const failed = ref(false);
+const busy = ref<string | null>(null);
+
+const rules = computed<RuleProposals[]>(() => data.value?.rules ?? []);
+
+async function load() {
+  try {
+    data.value = await getProposals();
+    failed.value = false;
+  } catch {
+    failed.value = true;
+  }
+}
+
+function key(rule: RuleProposals, p: MomentProposal): string {
+  return `${rule.id}:${p.moment}`;
+}
+
+function sourceLabel(p: MomentProposal): string {
+  if (p.source === "signal") {
+    return `opened just after this moment in ${p.evidence.situations} sessions`;
+  }
+  return "proposed from the rule's text";
+}
+
+async function decide(rule: RuleProposals, p: MomentProposal, verdict: Verdict) {
+  busy.value = key(rule, p);
+  try {
+    const out = await judgeProposals([{
+      rule_id: rule.id, moment: p.moment, verdict,
+      note: verdict === "confirm" ? "Mounted in Settings." : "Declined in Settings.",
+    }]);
+    if (out.refused.length) {
+      toast.show(out.refused[0].error, "error");
+      return;
+    }
+    await load();
+    // A mount changes the per-moment counts the list below shows.
+    if (verdict === "confirm") await store.load(true);
+  } catch (e) {
+    toast.show(apiErrorMessage(e, "Could not record that"), "error");
+  } finally {
+    busy.value = null;
+  }
+}
+
+onMounted(load);
+</script>
+
+<template>
+  <section v-if="failed || rules.length" class="proposals" aria-labelledby="proposals-title">
+    <h3 id="proposals-title" class="sub-title">Waiting on you</h3>
+    <p v-if="failed" class="error-msg">
+      The proposals could not be loaded.
+      <button type="button" class="btn-text" @click="load">Try again</button>
+    </p>
+    <template v-else>
+      <p class="field-hint">
+        Moments these rules may belong on. Mounting one makes the rule arrive whenever that
+        moment happens, whatever the work is about; declining keeps it from being proposed again.
+      </p>
+      <ul class="proposal-list">
+        <li v-for="rule in rules" :key="rule.id" class="proposal-rule">
+          <div class="rule-head">
+            <span class="rule-title">#{{ rule.id }} {{ rule.title }}</span>
+            <span v-if="rule.mounted.length" class="rule-mounted">
+              on <code v-for="m in rule.mounted" :key="m" class="moment-name">{{ m }}</code>
+            </span>
+          </div>
+          <p class="rule-statement">{{ rule.statement }}</p>
+          <ul class="proposal-items">
+            <li v-for="p in rule.proposals" :key="key(rule, p)" class="proposal-item">
+              <code class="moment-name">{{ p.moment }}</code>
+              <span class="proposal-why">
+                {{ p.why || sourceLabel(p) }}
+                <span v-if="p.why" class="proposal-source">· {{ sourceLabel(p) }}</span>
+              </span>
+              <span class="proposal-actions">
+                <button
+                  type="button" class="btn-primary btn-sm"
+                  :disabled="busy !== null"
+                  @click="decide(rule, p, 'confirm')"
+                >Mount</button>
+                <button
+                  type="button" class="btn-text"
+                  :disabled="busy !== null"
+                  @click="decide(rule, p, 'reject')"
+                >Not this</button>
+              </span>
+            </li>
+          </ul>
+        </li>
+      </ul>
+    </template>
+  </section>
+</template>
+
+<style scoped>
+.proposals { display: flex; flex-direction: column; gap: var(--fs-space-2); }
+.sub-title { margin: 0; font-size: 0.95rem; color: var(--fs-text-primary); }
+.proposal-list { list-style: none; margin: 0; padding: 0; display: flex; flex-direction: column; gap: var(--fs-space-2); }
+.proposal-rule {
+  padding: var(--fs-space-3);
+  background: var(--fs-surface-page);
+  border: 1px dashed var(--fs-border-color);
+  border-radius: var(--fs-radius-md);
+}
+.rule-head { display: flex; flex-wrap: wrap; align-items: baseline; gap: var(--fs-space-2); }
+.rule-title { font-size: var(--fs-size-body-sm); color: var(--fs-text-primary); font-weight: var(--fs-weight-medium); }
+.rule-mounted { display: inline-flex; flex-wrap: wrap; gap: var(--fs-space-1); font-size: var(--fs-size-tiny); color: var(--fs-text-tertiary); }
+.rule-statement { margin: var(--fs-space-1) 0 var(--fs-space-2); font-size: var(--fs-size-body-sm); color: var(--fs-text-secondary); line-height: var(--fs-leading-body); }
+.proposal-items { list-style: none; margin: 0; padding: 0; display: flex; flex-direction: column; gap: var(--fs-space-2); }
+.proposal-item { display: flex; flex-wrap: wrap; align-items: baseline; gap: var(--fs-space-2); font-size: var(--fs-size-body-sm); }
+.proposal-why { flex: 1 1 16rem; color: var(--fs-text-secondary); }
+.proposal-source { color: var(--fs-text-tertiary); font-size: var(--fs-size-tiny); }
+.proposal-actions { display: inline-flex; gap: var(--fs-space-2); margin-left: auto; }
+</style>
+
+<style src="@/assets/moments-shared.css" />
diff --git a/frontend/src/components/MomentsSettings.vue b/frontend/src/components/MomentsSettings.vue
index 80e89ce1..033e1194 100644
--- a/frontend/src/components/MomentsSettings.vue
+++ b/frontend/src/components/MomentsSettings.vue
@@ -1,6 +1,7 @@
 <script setup lang="ts">
 import { computed, onMounted, ref } from "vue";
 import { useMomentsStore } from "@/stores/moments";
+import MomentProposals from "@/components/MomentProposals.vue";
 import type { MomentAction, MomentUsage } from "@/api/moments";
 import { fmtDate } from "@/utils/dateFormat";
 
@@ -123,6 +124,8 @@ onMounted(() => store.load(true));
         The mount counts could not be read, so they are missing below — not zero.
       </p>
 
+      <MomentProposals />
+
       <ul class="moment-list">
         <li v-for="m in catalog" :key="m.name" class="moment-row">
           <div class="moment-head">
diff --git a/plugin/.claude-plugin/plugin.json b/plugin/.claude-plugin/plugin.json
index f083d771..c8493faf 100644
--- a/plugin/.claude-plugin/plugin.json
+++ b/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.05.1823",
+  "version": "2026.10.05.2003",
   "author": {
     "name": "Bryan Van Deusen"
   },
diff --git a/plugin/hooks/scribe_moment.sh b/plugin/hooks/scribe_moment.sh
index 1deafea9..0568fad4 100644
--- a/plugin/hooks/scribe_moment.sh
+++ b/plugin/hooks/scribe_moment.sh
@@ -16,10 +16,10 @@
 # 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.
+# SCRIBE'S OWN TOOLS ARE SKIPPED for delivery. 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. They are still written to the acts ledger below.
 #
 # 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
@@ -42,10 +42,6 @@ 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.
@@ -56,16 +52,48 @@ 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._-' '_')
 
+# Its own directory, outside SCRIBE_LEDGER_DIRS, so a compaction does not
+# sweep it: what is kept here describes the install and what the session
+# DID, not what its context holds. scribe_session_end.sh removes it.
+cache_dir="${TMPDIR:-/tmp}/scribe-moment"
+mkdir -p "$cache_dir" 2>/dev/null || true
+now=$(date +%s 2>/dev/null) || now=0
+
+# ── The acts ledger (milestone 458 step 7) ────────────────────────────────
+#
+# Every call, before any filter: `epoch<TAB>event`, one line each. When the
+# session then opens a rule, scribe_record_opened.sh sends the last few
+# minutes of these, and the server counts the open as evidence the rule
+# belongs on the moments they reached — which is how a rule nobody mounted
+# gets proposed for one. Recorded whether or not anything is mounted, since
+# an unmounted moment is exactly what the evidence is for, and including
+# Scribe's own tools (closing a task is a moment too). The open itself is
+# not an act. A large event (a file write) is kept as its tool name alone.
+if [ "$now" -gt 0 ] && [ "$tool_key" != "get_rule" ]; then
+  acts_file="$cache_dir/${safe_sid}.acts"
+  act=$(printf '%s' "$event" | tr -d '\n\r')
+  if [ "${#act}" -gt 16384 ]; then
+    esc=$(printf '%s' "$tool_name" | scribe_json_escape) || esc=""
+    act="{\"tool_name\":\"${esc}\",\"tool_input\":{}}"
+  fi
+  printf '%s\t%s\n' "$now" "$act" >> "$acts_file" 2>/dev/null || true
+  # Kept short: only the last few minutes are ever read.
+  size=$(wc -c < "$acts_file" 2>/dev/null) || size=0
+  if [ "${size:-0}" -gt 262144 ]; then
+    tail -n 50 "$acts_file" > "$acts_file.tmp" 2>/dev/null \
+      && mv -f "$acts_file.tmp" "$acts_file" 2>/dev/null
+  fi
+fi
+
+case "$tool_name" in
+  mcp__*scribe*__*) exit 0 ;;
+esac
+
 # ── 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.
-# Its own directory, outside SCRIBE_LEDGER_DIRS, so a compaction does not
-# sweep it: it describes the install, not what this context holds.
-cache_dir="${TMPDIR:-/tmp}/scribe-moment"
-mkdir -p "$cache_dir" 2>/dev/null || true
 tools_file="$cache_dir/${safe_sid}.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
diff --git a/plugin/hooks/scribe_record_opened.sh b/plugin/hooks/scribe_record_opened.sh
index 99359027..8dec556b 100644
--- a/plugin/hooks/scribe_record_opened.sh
+++ b/plugin/hooks/scribe_record_opened.sh
@@ -33,6 +33,10 @@
 # claim it supports is only ever "you opened this, pull it again if you no
 # longer hold it", which stays true in every case and carries its own remedy.
 #
+# SINCE MILESTONE 458 STEP 7 it also reports the open, with the calls that
+# came just before it, so the server can learn which moments a rule belongs
+# on (below). Silent on outage, like every arm that fires per call.
+#
 # EXIT 0, ALWAYS. This decorates a ledger; a bookkeeping failure must never
 # turn a successful tool call into a hook error. Worst case the id is missed
 # and the reader is offered a rule it already read — the cost of a wrong guess
@@ -68,4 +72,35 @@ safe_sid=$(printf '%s' "$session_id" | tr -c 'A-Za-z0-9._-' '_')
 # Stamped and append-only, exactly like the naming ledger — so the same reader
 # (`scribe_rules_live`) ages both, and the last entry for an id wins.
 printf '%s\n' "$rule_id" | scribe_rules_append "$state_dir/${safe_sid}.opened.ids"
+
+# ── What came just before the open (milestone 458 step 7) ─────────────────
+#
+# scribe_moment.sh writes every call to `scribe-moment/<sid>.acts`. The ones
+# from the last three minutes go to the server with this rule's id: a rule
+# opened just after a moment fired is evidence it belongs on that moment, and
+# once that repeats across sessions the server answers with one line asking
+# the reader to offer the operator the mount. Nothing is mounted by this.
+command -v curl >/dev/null 2>&1 || exit 0
+scribe_config || exit 0
+acts_file="${TMPDIR:-/tmp}/scribe-moment/${safe_sid}.acts"
+[ -f "$acts_file" ] || exit 0
+now=$(date +%s 2>/dev/null) || exit 0
+acts=$(awk -F '\t' -v since=$((now - 180)) \
+  '$1 + 0 >= since { sub(/^[^\t]*\t/, ""); print }' "$acts_file" 2>/dev/null \
+  | tail -n 30 | paste -sd ',' - 2>/dev/null)
+[ -n "$acts" ] || exit 0
+
+sid_esc=$(printf '%s' "$session_id" | scribe_json_escape) || exit 0
+event_cwd=$(scribe_json_pick "$event_flat" '.cwd')
+scope=$(scribe_scope_query "${event_cwd:-${CLAUDE_PROJECT_DIR:-$PWD}}")
+body=$(printf '{"rule_id":%s,"session_id":"%s","acts":[%s]}' "$rule_id" "$sid_esc" "$acts" \
+  | curl -fsS --max-time 3 \
+    -H "Authorization: Bearer ${token}" \
+    -H "Content-Type: application/json" \
+    --data-binary @- \
+    "${url%/}/api/plugin/rule-opened${scope:+?$scope}" 2>/dev/null) || exit 0
+
+context=$(scribe_json_pick "$(printf '%s' "$body" | scribe_json_flat)" '.context')
+[ -n "$context" ] || exit 0
+scribe_json_out PostToolUse "$context"
 exit 0
diff --git a/plugin/hooks/scribe_session_end.sh b/plugin/hooks/scribe_session_end.sh
index db216956..1fa8e0a7 100755
--- a/plugin/hooks/scribe_session_end.sh
+++ b/plugin/hooks/scribe_session_end.sh
@@ -24,18 +24,24 @@ set -uo pipefail
 # shellcheck source=plugin/hooks/scribe_defs.sh
 . "$(dirname "${BASH_SOURCE[0]}")/scribe_defs.sh"
 
-command -v curl >/dev/null 2>&1 || exit 0
-scribe_config || exit 0
-
 event=$(cat 2>/dev/null || true)
 [ -n "$event" ] || exit 0
 
 event_flat=$(printf '%s' "$event" | scribe_json_flat)
+session_id=$(scribe_json_pick "$event_flat" '.session_id')
+[ -n "$session_id" ] || exit 0
+
+# The moment arm's per-session files (its tool cache and the acts ledger,
+# milestone 458) describe a session that is over, on any exit — a clear
+# included, since the next conversation is a new session id.
+safe_sid=$(printf '%s' "$session_id" | tr -c 'A-Za-z0-9._-' '_')
+rm -f "${TMPDIR:-/tmp}/scribe-moment/${safe_sid}".* 2>/dev/null || true
+
 reason=$(scribe_json_pick "$event_flat" '.reason')
 [ "$reason" = "clear" ] && exit 0
 
-session_id=$(scribe_json_pick "$event_flat" '.session_id')
-[ -n "$session_id" ] || exit 0
+command -v curl >/dev/null 2>&1 || exit 0
+scribe_config || exit 0
 
 sid_enc=$(printf '%s' "$session_id" | scribe_urlenc) || exit 0
 curl -fsS --max-time 4 \
diff --git a/src/scribe/mcp/server.py b/src/scribe/mcp/server.py
index 2585d8db..5b1d5ce2 100644
--- a/src/scribe/mcp/server.py
+++ b/src/scribe/mcp/server.py
@@ -159,6 +159,10 @@ _READ_ONLY_TOOLS = frozenset({
     # retrieval_telemetry's reason, and needed by a read key so that a line
     # naming a moment can be understood by whoever was shown it.
     "list_moments",
+    # The pass over the corpus and its queue (milestone 458 step 7): which
+    # rules are unjudged and which proposals wait. Reads of the caller's own
+    # rules, as list_rules is.
+    "rules_to_mount", "rule_moment_proposals",
 })
 
 # Every tool that WRITES, by name. Nothing reads this set at runtime — a tool
@@ -210,6 +214,9 @@ _WRITE_TOOLS = frozenset({
     # Which actions reach which moment on this install (milestone 458). Each
     # changes what fires for every later session, so a read key is refused.
     "map_action", "unmap_action",
+    # Proposing moments for a rule writes a judgment row; judging one mounts
+    # or unmounts the rule (step 7).
+    "propose_rule_moments", "judge_rule_moments",
     # A reviewer's verdicts on logged menu lines (#4772) — rows carrying free
     # prose the agent authored, `rule_outcome`'s reason for being a write.
     "judge_menu",
diff --git a/src/scribe/mcp/tools/moments.py b/src/scribe/mcp/tools/moments.py
index e92df37e..731a6ade 100644
--- a/src/scribe/mcp/tools/moments.py
+++ b/src/scribe/mcp/tools/moments.py
@@ -15,6 +15,7 @@ from __future__ import annotations
 from scribe.mcp._context import current_user_id
 from scribe.services import moment_actions as actions_svc
 from scribe.services import moments as moments_svc
+from scribe.services import rule_moment_judgments as judgments_svc
 
 
 async def list_moments() -> dict:
@@ -105,7 +106,73 @@ async def unmap_action(tool: str, moment: str, match: str = "", reason: str = ""
     )
 
 
+async def rules_to_mount(limit: int = 25, offset: int = 0) -> dict:
+    """The rules nobody has decided the moments of — the pass over the corpus, a page at a time.
+
+    A rule written before moments existed is mounted on nothing and arrives
+    only when its words resemble the work. Read each rule here, decide WHEN it
+    applies, and record the answer with `propose_rule_moments`: the moments it
+    belongs on, or that none fits because it is about WHAT is done rather than
+    when. A rule leaves this list as soon as any answer is recorded, so the
+    next call returns the next unread rules.
+
+    Returns `rules` (id, title, kind, statement, when_to_apply, home),
+    `total_unjudged`, and the `catalog` to choose from.
+    """
+    return await judgments_svc.unjudged_rules(current_user_id(), limit=limit, offset=offset)
+
+
+async def propose_rule_moments(proposals: list[dict]) -> dict:
+    """Propose the moments rules belong on — for the operator to confirm, never mounted by this call.
+
+    Each item is either
+    `{"rule_id": N, "moments": ["work.finish", "reply.report"], "why": "…"}`
+    or `{"rule_id": N, "none": "why no moment fits"}`.
+
+    Propose a moment when the rule governs that point in the work whatever
+    the work is about — a rule about when work counts as done belongs where
+    work is finished, delivered, verified and reported. Say none fits when
+    the rule is about a subject (a library, a file, a style) and is best
+    reached by meaning. `why` is shown to the operator beside the proposal;
+    write it so they can say yes or no without opening the rule.
+
+    A moment proposal waits as `suggested` until judged; show the operator
+    what you proposed (`rule_moment_proposals`) and let them decide. A pair
+    already mounted or judged is skipped and listed under `skipped`; a bad
+    item is refused alone under `refused`.
+    """
+    return await judgments_svc.propose(current_user_id(), proposals)
+
+
+async def rule_moment_proposals(rule_id: int = 0) -> dict:
+    """The moment proposals waiting on the operator, grouped by rule.
+
+    Each rule carries what it is mounted on now and its proposals: the moment,
+    where the proposal came from (`pass` — read from the rule; `signal` — the
+    rule kept being opened just after that moment fired), the reason, and the
+    evidence counts. `rule_id` narrows to one rule.
+    """
+    return await judgments_svc.pending(current_user_id(), rule_id=rule_id or None)
+
+
+async def judge_rule_moments(judgments: list[dict]) -> dict:
+    """Confirm or reject moment proposals — on the operator's word, since a confirm mounts the rule.
+
+    Each item is `{"rule_id": N, "moment": "work.finish", "verdict":
+    "confirm" | "reject", "note": "why"}`. Confirm mounts the rule on that
+    moment beside what it already has; reject records that it does not belong
+    there (and unmounts it if it was mounted), so neither the pass nor the
+    signal proposes the pair again. Moment `""` judges a "no moment fits"
+    answer. Put the operator's reason in `note`.
+    """
+    return await judgments_svc.judge(current_user_id(), judgments)
+
+
 def register(mcp) -> None:
     mcp.tool(name="list_moments")(list_moments)
     mcp.tool(name="map_action")(map_action)
     mcp.tool(name="unmap_action")(unmap_action)
+    mcp.tool(name="rules_to_mount")(rules_to_mount)
+    mcp.tool(name="propose_rule_moments")(propose_rule_moments)
+    mcp.tool(name="rule_moment_proposals")(rule_moment_proposals)
+    mcp.tool(name="judge_rule_moments")(judge_rule_moments)
diff --git a/src/scribe/models/__init__.py b/src/scribe/models/__init__.py
index 5ba483eb..071c188a 100644
--- a/src/scribe/models/__init__.py
+++ b/src/scribe/models/__init__.py
@@ -82,6 +82,8 @@ from scribe.models.rulebook import (  # noqa: E402, F401
 )
 # After notes and rules: it foreign-keys both (milestone 440).
 from scribe.models.lesson_rule_link import LessonNoRule, LessonRuleLink  # noqa: E402, F401
+# After rules: it foreign-keys them (milestone 458 step 7).
+from scribe.models.rule_moment_judgment import RuleMomentJudgment  # noqa: E402, F401
 from scribe.models.repo_binding import RepoBinding  # noqa: E402, F401
 from scribe.models.forge_connection import ForgeConnection  # noqa: E402, F401
 from scribe.models.code_shape import CodeShape, CodeShapeConsumer, CodeShapeEvent, CodeShapeUse  # noqa: E402, F401
diff --git a/src/scribe/models/rule_moment_judgment.py b/src/scribe/models/rule_moment_judgment.py
new file mode 100644
index 00000000..2f83721e
--- /dev/null
+++ b/src/scribe/models/rule_moment_judgment.py
@@ -0,0 +1,81 @@
+from datetime import datetime
+
+from sqlalchemy import BigInteger, DateTime, ForeignKey, Text, UniqueConstraint
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column
+
+from scribe.models import Base
+from scribe.models.base import CreatedAtMixin, iso
+from scribe.models.lesson_rule_link import CONFIRMED, LINK_STATES, REJECTED, SUGGESTED
+
+# The same three states as a lesson's link to a rule, for the same reasons —
+# re-exported so a reader of this table need not know where they were first
+# named. CHECK ck_rule_moment_judgments_state (migration 0118, rule 36).
+JUDGMENT_STATES = LINK_STATES
+__all__ = [
+    "CONFIRMED", "JUDGMENT_STATES", "NO_MOMENT", "REJECTED", "SUGGESTED",
+    "SOURCES", "RuleMomentJudgment",
+]
+
+# The moment recorded for "this rule is about WHAT, not WHEN": no moment in
+# the catalog is when it applies, and it is right to reach it by meaning. A
+# row rather than an absence, for the reason LessonNoRule is one — a rule
+# looked at and found to have no moment must not read as one nobody looked at,
+# or every pass over the corpus would propose it again.
+NO_MOMENT = ""
+
+# Where a judgment came from. `pass` — an agent read the rule and proposed;
+# `signal` — the rule kept being opened just after a moment fired; `edit` — a
+# person changed the rule's moments directly, which is a judgment too.
+SOURCES = ("pass", "signal", "edit")
+
+
+class RuleMomentJudgment(Base, CreatedAtMixin):
+    """Whether a rule belongs on a moment, and who said so (milestone 458 step 7).
+
+    A mount (`rule_moments`) is the current answer; this is the reasoning
+    behind it and the answers that are NOT mounts. One row per (rule, moment):
+
+    - ``suggested`` — proposed by a pass, or by repeated opens after the
+      moment fired. Delivers nothing: a suggestion that mounted itself would
+      manufacture the opens it counts.
+    - ``confirmed`` — a judgment put the rule on the moment. The mount exists.
+    - ``rejected`` — a judgment said it does not belong there, kept with its
+      reason so the pair is never proposed again, by either source.
+
+    Moment ``""`` (NO_MOMENT), confirmed, is "no moment fits this rule".
+    """
+
+    __tablename__ = "rule_moment_judgments"
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
+    rule_id: Mapped[int] = mapped_column(
+        BigInteger, ForeignKey("rules.id", ondelete="CASCADE"), index=True,
+    )
+    moment: Mapped[str] = mapped_column(Text)
+    state: Mapped[str] = mapped_column(Text, default=SUGGESTED, server_default=SUGGESTED)
+    source: Mapped[str] = mapped_column(Text, default="pass", server_default="pass")
+    # Why — for a suggestion, what the proposer read in the rule; for a
+    # judgment, why it was confirmed or rejected.
+    note: Mapped[str | None] = mapped_column(Text, nullable=True)
+    # The co-occurrence evidence (signal source), in lesson_rules' shape.
+    evidence: Mapped[dict | None] = mapped_column(JSONB, nullable=True)
+    judged_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True,
+    )
+
+    __table_args__ = (
+        UniqueConstraint("rule_id", "moment", name="uq_rule_moment_judgments_pair"),
+    )
+
+    def to_dict(self) -> dict:
+        return {
+            "rule_id": self.rule_id,
+            "moment": self.moment,
+            "state": self.state,
+            "source": self.source,
+            "note": self.note or "",
+            "evidence": self.evidence or {},
+            "judged_at": iso(self.judged_at),
+            "created_at": iso(self.created_at),
+        }
diff --git a/src/scribe/routes/plugin.py b/src/scribe/routes/plugin.py
index b963b4de..dffa0f53 100644
--- a/src/scribe/routes/plugin.py
+++ b/src/scribe/routes/plugin.py
@@ -17,6 +17,7 @@ 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
+from scribe.services import rule_moment_judgments as judgments_svc
 from scribe.services import shape_check as shape_check_svc
 from scribe.services import task_claims as task_claims_svc
 from scribe.services.embeddings import memoized_query_embeddings
@@ -294,6 +295,35 @@ async def moment():
     })
 
 
+@plugin_bp.post("/rule-opened")
+@login_required
+async def rule_opened():
+    """A session opened a rule shortly after these acts (milestone 458 step 7).
+
+    Body: `{"rule_id", "session_id", "acts": [<PreToolUse event>, ...]}` —
+    the calls the plugin saw in the window before the `get_rule`. Each is
+    resolved to its moments, and the open is counted as evidence that the
+    rule belongs on them; once a pair has repeated across enough distinct
+    sessions, `context` carries the line asking the reader to offer the
+    mount. Nothing is mounted here. `repo` / `project_id` scope the evidence.
+    """
+    data = await request.get_json(silent=True) or {}
+    if not isinstance(data, dict):
+        return jsonify({"moments": [], "context": ""})
+    try:
+        rule_id = int(data.get("rule_id") or 0)
+    except (TypeError, ValueError):
+        rule_id = 0
+    acts = data.get("acts")
+    if rule_id <= 0 or not isinstance(acts, list):
+        return jsonify({"moments": [], "context": ""})
+    project_id, _repo, _unbound = await _project_scope()
+    return jsonify(await judgments_svc.opened_after(
+        g.user.id, rule_id, acts, session_id=str(data.get("session_id") or ""),
+        project_id=project_id or None,
+    ))
+
+
 @plugin_bp.post("/reply-rules")
 @login_required
 @memoized_query_embeddings
diff --git a/src/scribe/routes/retrieval.py b/src/scribe/routes/retrieval.py
index f21d53c4..ad7ebee6 100644
--- a/src/scribe/routes/retrieval.py
+++ b/src/scribe/routes/retrieval.py
@@ -20,6 +20,7 @@ from quart import Blueprint, jsonify, request
 from scribe.auth import get_current_user_id, login_required
 from scribe.services import moment_actions as moment_actions_svc
 from scribe.services import moments as moments_svc
+from scribe.services import rule_moment_judgments as judgments_svc
 from scribe.services import rulebooks as rulebooks_svc
 from scribe.services.retrieval_telemetry import moment_usage
 from scribe.services.retrieval_tuning import set_dial, current_settings, tuning_history
@@ -177,3 +178,27 @@ async def map_action_route():
 @login_required
 async def unmap_action_route():
     return await _mapping_change(moment_actions_svc.unmap_action)
+
+
+@retrieval_bp.route("/moments/proposals", methods=["GET"])
+@login_required
+async def moment_proposals_route():
+    """The moment proposals waiting on a judgment (milestone 458 step 7) —
+    `rule_moment_proposals`' payload, from the same service. `?rule_id=`
+    narrows to one rule, for the rule editor."""
+    try:
+        rule_id = int(request.args.get("rule_id") or 0)
+    except ValueError:
+        return jsonify({"error": "rule_id must be a whole number"}), 400
+    return jsonify(await judgments_svc.pending(get_current_user_id(), rule_id=rule_id or None))
+
+
+@retrieval_bp.route("/moments/proposals/judge", methods=["POST"])
+@login_required
+async def judge_moment_proposals_route():
+    """Confirm or reject proposals: `{"judgments": [{rule_id, moment, verdict,
+    note}]}` — `judge_rule_moments` from the browser. A confirm mounts."""
+    data = await request.get_json(silent=True)
+    if not isinstance(data, dict) or not isinstance(data.get("judgments"), list):
+        return jsonify({"error": "Expected {\"judgments\": [...]}"}), 400
+    return jsonify(await judgments_svc.judge(get_current_user_id(), data["judgments"]))
diff --git a/src/scribe/services/backup.py b/src/scribe/services/backup.py
index c765b8cb..1bb8bfc4 100644
--- a/src/scribe/services/backup.py
+++ b/src/scribe/services/backup.py
@@ -21,6 +21,7 @@ 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.rule_moment_judgment import RuleMomentJudgment
 from scribe.models.code_shape import CodeShape, CodeShapeEvent, CodeShapeUse
 from scribe.models.project import Project
 from scribe.models.repo_binding import RepoBinding
@@ -104,8 +105,12 @@ logger = logging.getLogger(__name__)
 # 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.
+# v23 (2026-10) added rule_moment_judgments (milestone 458 step 7): the
+# proposals waiting on a mount, and the rejections and "no moment fits"
+# answers that stop a pass or the open-after-moment signal proposing the same
+# pair again. Losing them puts every answered question back on the list.
 # Bump when the serialized schema changes.
-BACKUP_VERSION = 22
+BACKUP_VERSION = 23
 
 # 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
@@ -154,6 +159,8 @@ _BACKED_UP = [
     "moment_mappings",
     # v22 (2026-10): the moments each rule arrives at (milestone 458).
     "rule_moments",
+    # v23 (2026-10): proposals and judgments about those mounts (step 7).
+    "rule_moment_judgments",
 ]
 
 # Tables intentionally NOT in the backup, surfaced in the payload so the gap is
@@ -253,6 +260,8 @@ _COLUMN_EXCLUSIONS: dict[str, set[str]] = {
     "rule_relations": {"id", "created_at"},
     # The pair is the row; everything else is the judgment and its evidence.
     "lesson_rule_links": {"id"},
+    # The same shape: the (rule, moment) pair is the row.
+    "rule_moment_judgments": {"id"},
     # Keyed by the lesson itself; nothing to exclude, every column travels.
     "lesson_no_rule": set(),
     "note_usage_events": {"id"},
@@ -346,6 +355,7 @@ _IMPORT_COLUMN_EXCLUSIONS: dict[str, set[str]] = {
     "note_supersessions": {"id", "created_at"},
     "rule_relations": {"id", "created_at"},
     "lesson_rule_links": {"id"},
+    "rule_moment_judgments": {"id"},
     "lesson_no_rule": set(),
     "note_usage_events": {"id"},
     "rule_usage_events": {"id"},
@@ -755,6 +765,20 @@ def _rule_moment_rows(rows) -> list[dict]:
     return [{"rule_id": rule_id, "moment": moment} for rule_id, moment in rows]
 
 
+def _rule_moment_judgment_rows(rows) -> list[dict]:
+    """Proposals and judgments about a rule's moments (v23). The rule id is a
+    SOURCE id, remapped at restore; the moment is a catalog name."""
+    return [
+        {
+            "rule_id": r.rule_id, "moment": r.moment, "state": r.state,
+            "source": r.source, "note": r.note, "evidence": r.evidence,
+            "judged_at": r.judged_at.isoformat() if r.judged_at else None,
+            "created_at": r.created_at.isoformat() if r.created_at else None,
+        }
+        for r 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."""
@@ -850,6 +874,9 @@ async def export_full_backup() -> dict:
         rule_moment_rows = (await session.execute(
             select(rule_moments_t.c.rule_id, rule_moments_t.c.moment)
         )).all()
+        rule_moment_judgments = (await session.execute(
+            select(RuleMomentJudgment)
+        )).scalars().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()
@@ -917,6 +944,7 @@ async def export_full_backup() -> dict:
         "canonical_systems": _canonical_system_rows(canonical_systems),
         "rule_systems": _rule_system_rows(rule_system_rows),
         "rule_moments": _rule_moment_rows(rule_moment_rows),
+        "rule_moment_judgments": _rule_moment_judgment_rows(rule_moment_judgments),
         "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),
@@ -1055,6 +1083,9 @@ async def export_user_backup(user_id: int) -> dict:
             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 []
+        rule_moment_judgments = (await session.execute(
+            select(RuleMomentJudgment).where(RuleMomentJudgment.rule_id.in_(_rule_ids))
+        )).scalars().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
@@ -1137,6 +1168,7 @@ async def export_user_backup(user_id: int) -> dict:
         "canonical_systems": _canonical_system_rows(canonical_systems),
         "rule_systems": _rule_system_rows(rule_system_rows),
         "rule_moments": _rule_moment_rows(rule_moment_rows),
+        "rule_moment_judgments": _rule_moment_judgment_rows(rule_moment_judgments),
         "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),
@@ -1535,6 +1567,25 @@ def _build_lesson_rule_link(row: dict, maps: _Maps) -> LessonRuleLink | None:
     )
 
 
+def _build_rule_moment_judgment(row: dict, maps: _Maps) -> RuleMomentJudgment | None:
+    """Skipped when its rule did not restore — the judgment is about that
+    rule, and a remap miss would hang it on whatever took the number."""
+    rule = maps.rules.get(row.get("rule_id", 0))
+    if rule is None or row.get("moment") is None:
+        return None
+    return RuleMomentJudgment(
+        rule_id=rule,
+        moment=row["moment"],
+        state=row.get("state") or "suggested",
+        source=row.get("source") or "pass",
+        note=row.get("note") or None,
+        evidence=row.get("evidence"),
+        # Absent stays absent: a suggestion was never judged.
+        judged_at=_dt_or_none(row.get("judged_at")),
+        created_at=_dt(row.get("created_at")),
+    )
+
+
 def _build_rule_version(row: dict, maps: _Maps) -> RuleVersion | None:
     rid = maps.rules.get(row.get("rule_id", 0))
     if rid is None:
@@ -1949,7 +2000,7 @@ 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_moments": 0,
+        "rule_systems": 0, "rule_moments": 0, "rule_moment_judgments": 0,
         "rule_relations": 0, "rule_versions": 0,
         "retrieval_tuning_events": 0, "lesson_rule_links": 0,
         "lesson_no_rule": 0, "moment_mappings": 0,
@@ -2160,6 +2211,15 @@ async def _restore_v2(data: dict) -> dict:
             ))
             stats["rule_moments"] += 1
 
+        # Proposals and judgments about those mounts (v23); archives before
+        # v23 carry none, and every rule restores unjudged.
+        for rj in data.get("rule_moment_judgments", []):
+            judgment = _build_rule_moment_judgment(rj, maps)
+            if judgment is None:
+                continue
+            session.add(judgment)
+            stats["rule_moment_judgments"] += 1
+
         for rr in data.get("rule_relations", []):
             relation = _build_rule_relation(rr, maps)
             if relation is None:
diff --git a/src/scribe/services/lesson_rules.py b/src/scribe/services/lesson_rules.py
index f3de5392..8c073f6f 100644
--- a/src/scribe/services/lesson_rules.py
+++ b/src/scribe/services/lesson_rules.py
@@ -485,16 +485,18 @@ async def attach_rule_lessons(user_id: int, data: dict, rule_id: int) -> None:
 def situation_key(arm: str, text: str) -> str:
     """A fingerprint of one situation, so a repeat counts once.
 
-    `arm` is part of the key — "p" for a prompt, "w" for a file being written
-    — because the two name different kinds of situation and must not collide.
+    `arm` is part of the key — "p" for a prompt, "w" for a file being written,
+    "s" for a session (the rule↔moment signal, milestone 458 step 7) —
+    because each names a different kind of situation and must not collide.
     On the prompt arm the text is the prompt: lowercased, split into word
     tokens of _MIN_TOKEN or more, de-duplicated and sorted, so the same ask
     re-sent with different spacing, punctuation, case or word order is one
     situation. On the write arm the text is the PATH, not the code: every
     edit to one file is one situation, however much the code differs between
-    them. Empty when there is nothing to key on.
+    them. The session arm keys on the session id as given. Empty when there
+    is nothing to key on.
     """
-    if arm == "w":
+    if arm in ("w", "s"):
         basis = (text or "").strip()
     else:
         tokens = sorted({t for t in re.findall(r"[a-z0-9]+", (text or "").lower())
diff --git a/src/scribe/services/rule_moment_judgments.py b/src/scribe/services/rule_moment_judgments.py
new file mode 100644
index 00000000..3eced3b3
--- /dev/null
+++ b/src/scribe/services/rule_moment_judgments.py
@@ -0,0 +1,527 @@
+"""Which rules belong on which moments — proposals and judgments (milestone 458 step 7).
+
+A rule written before moments existed is mounted on nothing, and reaches a
+session only when its words resemble the work. Mounting the corpus is a
+judgment about each rule — WHEN does it apply? — and nothing in a rule's row
+says whether that judgment was ever made. This service records it, from two
+sources that both stop at a proposal:
+
+- THE PASS. An agent reads the rules nobody has judged (`unjudged_rules`),
+  proposes moments for each, or says none fits (`propose`). A proposal mounts
+  nothing; a person confirms it (`judge`).
+- THE SIGNAL. A rule a session OPENS shortly after a moment fired is evidence
+  it belongs on that moment. Recorded per distinct session with the evidence
+  model lesson↔rule soft links use (lesson_rules: three situations, a
+  cooldown), and once it crosses the bar the next open asks the reader to
+  offer the mount (`co_occurred`).
+
+Neither source mounts anything by itself. A suggestion that delivered its rule
+would manufacture the opens it counts, and the operator's corpus is theirs to
+mount.
+
+EDITS ARE JUDGMENTS TOO. A person changing a rule's moments directly says
+which moments it belongs on, so `record_mount_change` (called from the one
+write path, `rulebooks.set_rule_moments`) confirms what was added and rejects
+what was removed — a moment taken off a rule must not be proposed straight
+back by the signal.
+
+ACL (rule 78): rules are owner-scoped, and every read and write here goes
+through `rulebooks._owned_rules_clause` / `_fetch_owned_rule`.
+"""
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+
+from sqlalchemy import exists, func, not_, select
+from sqlalchemy.exc import IntegrityError
+
+from scribe.models import async_session
+from scribe.models.rule_moment_judgment import (
+    CONFIRMED, NO_MOMENT, REJECTED, SUGGESTED, RuleMomentJudgment,
+)
+from scribe.models.rulebook import Rule, rule_moments as rule_moments_t
+from scribe.services import lesson_rules
+from scribe.services import moments as moments_svc
+
+logger = logging.getLogger(__name__)
+
+# What a caller says to judge one pair — acts, not the states they produce.
+VERDICTS = {"confirm": CONFIRMED, "reject": REJECTED}
+
+# Recorded on the rows an edit wrote, so a reader can tell a direct change from
+# an accepted proposal.
+_MOUNTED_NOTE = "mounted by an edit"
+_UNMOUNTED_NOTE = "unmounted by an edit"
+# On a "no moment fits" answer a later mount overturned.
+_OVERTURNED_NOTE = "a moment was mounted after all"
+
+# How many unjudged rules one pass page hands over. Enough to batch the
+# reading, few enough that every rule is actually read.
+PASS_PAGE = 25
+# How long after a moment an open still counts as following it. Long enough
+# for a session to see a line, decide it matters and open the rule; short
+# enough that the moment is still what the work is doing.
+SIGNAL_WINDOW_SECONDS = 180
+# Moments too ubiquitous to say anything about the rule opened after them:
+# nearly every open follows a command or a change, so counting them would
+# propose every rule onto both.
+SIGNAL_SKIP = frozenset({"work.run", "work.change"})
+
+
+def _clean_moment(name) -> str:
+    """A catalog moment, normalised; "" stays "" (no moment fits)."""
+    raw = (name or "").strip() if isinstance(name, str) else ""
+    if raw == NO_MOMENT:
+        return NO_MOMENT
+    return moments_svc.require_moment(raw)
+
+
+def _int(v) -> int | None:
+    try:
+        i = int(v)
+    except (TypeError, ValueError):
+        return None
+    return i if i > 0 else None
+
+
+async def _owned_rules(session, user_id: int, rule_ids) -> dict[int, Rule]:
+    from scribe.services.rulebooks import _owned_rules_clause
+
+    ids = [i for i in (_int(r) for r in rule_ids or []) if i]
+    if not ids:
+        return {}
+    rows = (await session.execute(
+        select(Rule).where(Rule.id.in_(ids)).where(_owned_rules_clause(user_id))
+    )).scalars().all()
+    return {r.id: r for r in rows}
+
+
+async def _mounts(session, rule_ids) -> dict[int, set[str]]:
+    if not rule_ids:
+        return {}
+    rows = (await session.execute(
+        select(rule_moments_t.c.rule_id, rule_moments_t.c.moment)
+        .where(rule_moments_t.c.rule_id.in_(list(rule_ids)))
+    )).all()
+    out: dict[int, set[str]] = {}
+    for rid, moment in rows:
+        out.setdefault(rid, set()).add(moment)
+    return out
+
+
+async def _rows(session, rule_ids) -> dict[tuple[int, str], RuleMomentJudgment]:
+    if not rule_ids:
+        return {}
+    rows = (await session.execute(
+        select(RuleMomentJudgment).where(RuleMomentJudgment.rule_id.in_(list(rule_ids)))
+    )).scalars().all()
+    return {(r.rule_id, r.moment): r for r in rows}
+
+
+def _brief(rule: Rule) -> dict:
+    return {
+        "id": rule.id,
+        "title": rule.title,
+        "kind": rule.kind or "rule",
+        "statement": rule.statement,
+        "when_to_apply": rule.when_to_apply or "",
+        "home": "project" if rule.project_id else "global",
+    }
+
+
+# ── The pass ─────────────────────────────────────────────────────────────────
+
+
+def _unjudged_clause():
+    """No mount and no judgment row of any kind — a rule nobody has looked at."""
+    mounted = exists().where(rule_moments_t.c.rule_id == Rule.id)
+    judged = exists().where(RuleMomentJudgment.rule_id == Rule.id)
+    return not_(mounted), not_(judged)
+
+
+async def unjudged_rules(user_id: int, limit: int = PASS_PAGE, offset: int = 0) -> dict:
+    """The caller's rules with no moment and no judgment, a page at a time.
+
+    A rule leaves this list the moment any answer is recorded for it — a
+    proposal, a mount, or "no moment fits" — so a pass run in pieces, or by
+    two sessions, does not read the same rule twice.
+    """
+    from scribe.services.rulebooks import _owned_rules_clause
+
+    limit = max(1, min(int(limit or PASS_PAGE), 100))
+    offset = max(0, int(offset or 0))
+    clauses = (_owned_rules_clause(user_id), *_unjudged_clause())
+    async with async_session() as session:
+        total = (await session.execute(
+            select(func.count()).select_from(Rule).where(*clauses)
+        )).scalar_one()
+        rows = (await session.execute(
+            select(Rule).where(*clauses).order_by(Rule.id).limit(limit).offset(offset)
+        )).scalars().all()
+    return {
+        "rules": [_brief(r) for r in rows],
+        "total_unjudged": int(total),
+        "offset": offset,
+        "catalog": moments_svc.catalog(),
+    }
+
+
+async def propose(user_id: int, proposals: list[dict]) -> dict:
+    """Record a pass's proposals: moments for a rule, or that none fits.
+
+    Each item is `{rule_id, moments: [...], why}` or `{rule_id, none: "why"}`.
+    A moment proposal lands `suggested` and mounts nothing. "No moment fits"
+    lands as the agent's own answer (confirmed, moment ""): it changes nothing
+    that surfaces, and it is what keeps the rule off the next pass — the
+    signal can still propose a moment for it later, on evidence.
+
+    A pair already mounted or already judged is skipped and said so; a pair
+    already suggested has its reason refreshed. A bad item is refused alone,
+    with why, and the rest are written.
+    """
+    now = datetime.now(timezone.utc)
+    proposed, no_moment = 0, 0
+    skipped: list[dict] = []
+    refused: list[dict] = []
+    async with async_session() as session:
+        owned = await _owned_rules(session, user_id, [p.get("rule_id") for p in proposals or []
+                                                      if isinstance(p, dict)])
+        mounts = await _mounts(session, list(owned))
+        existing = await _rows(session, list(owned))
+        for item in proposals or []:
+            if not isinstance(item, dict):
+                refused.append({"item": item, "error": "each proposal is an object"})
+                continue
+            rid = _int(item.get("rule_id"))
+            if rid is None or rid not in owned:
+                refused.append({"rule_id": item.get("rule_id"),
+                                "error": "not a rule you own"})
+                continue
+            none_why = (item.get("none") or "").strip() if isinstance(item.get("none"), str) else ""
+            names = item.get("moments")
+            if none_why and names:
+                refused.append({"rule_id": rid, "error": (
+                    "moments and none are the two answers to \"when does this rule "
+                    "apply?\" — give the one that holds")})
+                continue
+            if none_why:
+                if mounts.get(rid):
+                    skipped.append({"rule_id": rid, "moment": NO_MOMENT,
+                                    "why": "already mounted on " + ", ".join(sorted(mounts[rid]))})
+                    continue
+                row = existing.get((rid, NO_MOMENT))
+                if row is None:
+                    row = RuleMomentJudgment(rule_id=rid, moment=NO_MOMENT, source="pass")
+                    session.add(row)
+                    existing[(rid, NO_MOMENT)] = row
+                row.state, row.note, row.judged_at = CONFIRMED, none_why, now
+                no_moment += 1
+                continue
+            try:
+                wanted = moments_svc.require_moments(names if names is not None else []) or []
+            except ValueError as exc:
+                refused.append({"rule_id": rid, "error": str(exc)})
+                continue
+            if not wanted:
+                refused.append({"rule_id": rid, "error": (
+                    "name at least one moment, or say none fits with none=\"why\"")})
+                continue
+            why = (item.get("why") or "").strip() if isinstance(item.get("why"), str) else ""
+            for moment in wanted:
+                if moment in mounts.get(rid, set()):
+                    skipped.append({"rule_id": rid, "moment": moment, "why": "already mounted"})
+                    continue
+                row = existing.get((rid, moment))
+                if row is not None and row.state != SUGGESTED:
+                    skipped.append({"rule_id": rid, "moment": moment,
+                                    "why": f"already {row.state}" + (f": {row.note}" if row.note else "")})
+                    continue
+                if row is None:
+                    row = RuleMomentJudgment(rule_id=rid, moment=moment, state=SUGGESTED, source="pass")
+                    session.add(row)
+                    existing[(rid, moment)] = row
+                row.note = why or row.note
+                proposed += 1
+        await session.commit()
+    return {"proposed": proposed, "no_moment": no_moment,
+            "skipped": skipped, "refused": refused}
+
+
+async def pending(user_id: int, rule_id: int | None = None, limit: int = 200) -> dict:
+    """The proposals waiting on a judgment, grouped by rule.
+
+    Each rule carries what it is mounted on now, so a proposal is read beside
+    the mounts it would add to.
+    """
+    from scribe.services.rulebooks import _owned_rules_clause
+
+    limit = max(1, min(int(limit or 200), 500))
+    async with async_session() as session:
+        stmt = (
+            select(RuleMomentJudgment, Rule)
+            .join(Rule, Rule.id == RuleMomentJudgment.rule_id)
+            .where(RuleMomentJudgment.state == SUGGESTED, _owned_rules_clause(user_id))
+        )
+        if rule_id:
+            stmt = stmt.where(Rule.id == int(rule_id))
+        rows = (await session.execute(
+            stmt.order_by(Rule.id, RuleMomentJudgment.moment).limit(limit)
+        )).all()
+        mounts = await _mounts(session, {r.id for _, r in rows})
+    order = {name: i for i, name in enumerate(moments_svc.MOMENTS)}
+    grouped: dict[int, dict] = {}
+    for judgment, rule in rows:
+        entry = grouped.setdefault(rule.id, {
+            **_brief(rule),
+            "mounted": sorted(mounts.get(rule.id, set()),
+                              key=lambda n: (order.get(n, len(order)), n)),
+            "proposals": [],
+        })
+        entry["proposals"].append({
+            "moment": judgment.moment,
+            "source": judgment.source,
+            "why": judgment.note or "",
+            "evidence": lesson_rules.evidence_summary(judgment.evidence),
+            "created_at": judgment.created_at.isoformat() if judgment.created_at else None,
+        })
+    rules = list(grouped.values())
+    return {"rules": rules, "total": sum(len(r["proposals"]) for r in rules)}
+
+
+async def judge(user_id: int, judgments: list[dict]) -> dict:
+    """Confirm or reject proposals — or any (rule, moment) pair, proposed or not.
+
+    Confirm MOUNTS the rule on the moment (alongside what it already has);
+    reject records that it does not belong there and, if it was mounted,
+    unmounts it. Both go through `rulebooks.set_rule_moments`, the one write
+    path, which records the judgment with this item's `note`. A moment of ""
+    judges the "no moment fits" answer itself. A bad item is refused alone.
+    """
+    from scribe.services.rulebooks import list_rule_moments, set_rule_moments
+
+    done: list[dict] = []
+    refused: list[dict] = []
+    for item in judgments or []:
+        if not isinstance(item, dict):
+            refused.append({"item": item, "error": "each judgment is an object"})
+            continue
+        rid = _int(item.get("rule_id"))
+        state = VERDICTS.get(str(item.get("verdict") or "").strip().lower())
+        if rid is None or state is None:
+            refused.append({"rule_id": item.get("rule_id"), "moment": item.get("moment"),
+                            "error": f"needs rule_id and a verdict, one of {sorted(VERDICTS)}"})
+            continue
+        try:
+            moment = _clean_moment(item.get("moment"))
+        except ValueError as exc:
+            refused.append({"rule_id": rid, "error": str(exc)})
+            continue
+        note = (item.get("note") or "").strip() if isinstance(item.get("note"), str) else ""
+        if moment == NO_MOMENT:
+            ok = await _judge_no_moment(user_id, rid, state, note)
+        else:
+            current = (await list_rule_moments([rid])).get(rid, [])
+            after = (current + [moment] if state == CONFIRMED and moment not in current
+                     else [m for m in current if m != moment] if state == REJECTED
+                     else current)
+            ok = await set_rule_moments(rid, user_id, after, note=note, judged=[moment],
+                                        verdict=state) is not None
+        if not ok:
+            refused.append({"rule_id": rid, "moment": moment, "error": "not a rule you own"})
+            continue
+        done.append({"rule_id": rid, "moment": moment, "state": state})
+    return {"judged": done, "refused": refused}
+
+
+async def _judge_no_moment(user_id: int, rule_id: int, state: str, note: str) -> bool:
+    now = datetime.now(timezone.utc)
+    async with async_session() as session:
+        if not await _owned_rules(session, user_id, [rule_id]):
+            return False
+        row = (await _rows(session, [rule_id])).get((rule_id, NO_MOMENT))
+        if row is None:
+            row = RuleMomentJudgment(rule_id=rule_id, moment=NO_MOMENT, source="edit")
+            session.add(row)
+        row.state, row.judged_at = state, now
+        row.note = note or row.note
+        await session.commit()
+    return True
+
+
+async def record_mount_change(
+    session, rule_id: int, before, after, *, note: str = "",
+    judged=(), verdict: str | None = None,
+) -> None:
+    """Record the judgments a change of mounts makes, in the caller's session.
+
+    Added moments are confirmed, removed ones rejected — a mount taken off
+    must not be proposed straight back. `judged` names moments a `judge`
+    call decided explicitly, so a rejection of a moment that was never
+    mounted is recorded too. A suggestion that was accepted keeps its source,
+    so the record still says where the idea came from. Any mount overturns a
+    "no moment fits" answer.
+    """
+    before, after = set(before or ()), set(after or ())
+    added = after - before
+    removed = before - after
+    explicit = {m for m in judged or () if m} - added - removed
+    if not (added or removed or explicit):
+        return
+    now = datetime.now(timezone.utc)
+    rows = await _rows(session, [rule_id])
+
+    def put(moment: str, state: str, default_note: str) -> None:
+        row = rows.get((rule_id, moment))
+        if row is None:
+            row = RuleMomentJudgment(rule_id=rule_id, moment=moment, source="edit")
+            session.add(row)
+            rows[(rule_id, moment)] = row
+        row.state, row.judged_at = state, now
+        row.note = note or default_note or row.note
+
+    for moment in added:
+        put(moment, CONFIRMED, _MOUNTED_NOTE)
+    for moment in removed:
+        put(moment, REJECTED, _UNMOUNTED_NOTE)
+    for moment in explicit:
+        if verdict in (CONFIRMED, REJECTED):
+            put(moment, verdict, "")
+    if after:
+        none_row = rows.get((rule_id, NO_MOMENT))
+        if none_row is not None and none_row.state == CONFIRMED:
+            none_row.state, none_row.judged_at = REJECTED, now
+            none_row.note = _OVERTURNED_NOTE
+
+
+# ── The signal ───────────────────────────────────────────────────────────────
+
+
+def _proposal_line(rule: Rule, moment: str, evidence: dict) -> str:
+    counts = lesson_rules.evidence_summary(evidence)
+    kind = rule.kind or "rule"
+    return (
+        f"> {kind.capitalize()} #{rule.id} \"{rule.title}\" has been opened just after "
+        f"`{moment}` in {counts['situations']} distinct sessions — it may belong on "
+        f"that moment, where it would arrive without being searched for. Offer the "
+        f"operator the mount in one line; on their yes, `judge_rule_moments("
+        f"[{{\"rule_id\": {rule.id}, \"moment\": \"{moment}\", \"verdict\": \"confirm\", "
+        f"\"note\": \"why\"}}])` mounts it; on a no, `\"reject\"` with their why stops "
+        f"the asking."
+    )
+
+
+def signal_moments(moments) -> list[str]:
+    """The moments an open may be counted against: catalog or procedure
+    moments, minus the ubiquitous ones, de-duplicated in order."""
+    out: list[str] = []
+    for m in moments or []:
+        name = (m or "").strip().lower() if isinstance(m, str) else ""
+        if name and name not in SIGNAL_SKIP and moments_svc.is_moment(name) and name not in out:
+            out.append(name)
+    return out
+
+
+async def co_occurred(
+    user_id: int, rule_id: int, moments, *, situation: str,
+    project_id: int | None = None,
+) -> str:
+    """Record that a rule was opened shortly after these moments fired, and
+    return a proposal line once a (rule, moment) pair has earned one.
+
+    `situation` is what makes two opens distinct — the session, so a rule
+    opened after every push in one long session counts once. A pair already
+    mounted or judged gathers nothing. One proposal per open at most. Fails
+    open: this rides a hook, and bookkeeping must never break the read.
+    """
+    try:
+        return await _co_occurred(user_id, rule_id, moments, situation, project_id)
+    except Exception:
+        logger.warning("rule/moment co-occurrence could not be recorded", exc_info=True)
+        return ""
+
+
+async def _co_occurred(user_id, rule_id, moments, situation, project_id) -> str:
+    wanted = signal_moments(moments)
+    key = lesson_rules.situation_key("s", situation)
+    rid = _int(rule_id)
+    if not wanted or not key or rid is None:
+        return ""
+    now = datetime.now(timezone.utc)
+    proposal = None
+    async with async_session() as session:
+        owned = await _owned_rules(session, user_id, [rid])
+        rule = owned.get(rid)
+        if rule is None:
+            return ""
+        mounted = (await _mounts(session, [rid])).get(rid, set())
+        existing = await _rows(session, [rid])
+        for moment in wanted:
+            if moment in mounted:
+                continue
+            row = existing.get((rid, moment))
+            if row is not None and row.state != SUGGESTED:
+                continue
+            if row is None:
+                row = RuleMomentJudgment(rule_id=rid, moment=moment, state=SUGGESTED, source="signal")
+                session.add(row)
+            ev = lesson_rules.add_evidence(row.evidence, key, project_id, now)
+            if proposal is None and lesson_rules.proposal_due(ev, now):
+                ev["proposed_at"] = now.isoformat()
+                ev["proposed_count"] = int(ev.get("proposed_count") or 0) + 1
+                proposal = (moment, ev)
+            row.evidence = ev
+        try:
+            await session.commit()
+        except IntegrityError:
+            # Two opens created the same pair at once; the other stands.
+            await session.rollback()
+            return ""
+    if proposal is None:
+        return ""
+    return _proposal_line(rule, *proposal)
+
+
+# The most acts one open is checked against. The hook sends only the window's
+# acts, so this bounds a misbehaving client rather than an ordinary session.
+_ACTS_CAP = 30
+
+
+async def opened_after(
+    user_id: int, rule_id: int, acts, *, session_id: str,
+    project_id: int | None = None,
+) -> dict:
+    """A rule was opened; `acts` are the tool calls that came just before it.
+
+    Each act is a PreToolUse event as the plugin recorded it (`tool_name`,
+    `tool_input`), resolved to its moments by the same mappings delivery
+    uses — so a moment counted here is one that fired, on this install, for
+    that call. Returns `moments` (what the window reached, after the skip
+    list) and `context` (a proposal line, or "").
+    """
+    from scribe.services.moment_actions import moments_for
+
+    reached: list[str] = []
+    for act in list(acts or [])[:_ACTS_CAP]:
+        if not isinstance(act, dict):
+            continue
+        tool = str(act.get("tool_name") or "").strip()
+        if not tool:
+            continue
+        tool_input = act.get("tool_input")
+        try:
+            hits = await moments_for(user_id, tool, tool_input if isinstance(tool_input, dict) else {})
+        except Exception:
+            logger.debug("acts before an open could not be resolved", exc_info=True)
+            continue
+        for hit in hits:
+            name = hit.get("moment") if isinstance(hit, dict) else None
+            if name and name not in reached:
+                reached.append(name)
+    moments = signal_moments(reached)
+    context = ""
+    if moments and (session_id or "").strip():
+        context = await co_occurred(
+            user_id, rule_id, moments, situation=session_id.strip(), project_id=project_id,
+        )
+    return {"moments": moments, "context": context}
diff --git a/src/scribe/services/rulebooks.py b/src/scribe/services/rulebooks.py
index 96bfa988..7c33b0ca 100644
--- a/src/scribe/services/rulebooks.py
+++ b/src/scribe/services/rulebooks.py
@@ -1111,7 +1111,8 @@ async def set_rule_systems(
 
 
 async def set_rule_moments(
-    rule_id: int, user_id: int, moments: list[str],
+    rule_id: int, user_id: int, moments: list[str], *,
+    note: str = "", judged=(), verdict: str | None = None,
 ) -> list[str] | None:
     """Replace which MOMENTS a rule arrives at (milestone 458). None if not owned.
 
@@ -1119,15 +1120,24 @@ async def set_rule_moments(
     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.
+
+    THE ONE WRITE PATH for mounts, so it is also where a change is recorded as
+    a judgment (step 7): what was added is confirmed, what was removed is
+    rejected, in the same transaction as the mounts. `note`, `judged` and
+    `verdict` come from `rule_moment_judgments.judge`, which says why.
     """
     from scribe.models.rulebook import rule_moments as rule_moments_t
     from scribe.services.moments import require_moments
+    from scribe.services.rule_moment_judgments import record_mount_change
 
     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
+        before = (await session.execute(
+            select(rule_moments_t.c.moment).where(rule_moments_t.c.rule_id == rule_id)
+        )).scalars().all()
         await session.execute(
             sql_delete(rule_moments_t).where(rule_moments_t.c.rule_id == rule_id)
         )
@@ -1135,6 +1145,9 @@ async def set_rule_moments(
             await session.execute(
                 insert(rule_moments_t).values(rule_id=rule_id, moment=moment)
             )
+        await record_mount_change(
+            session, rule_id, before, wanted, note=note, judged=judged, verdict=verdict,
+        )
         await session.commit()
         return wanted
 
diff --git a/tests/test_integration_rule_moment_judgments.py b/tests/test_integration_rule_moment_judgments.py
new file mode 100644
index 00000000..eba4f22c
--- /dev/null
+++ b/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"
diff --git a/tests/test_integration_rule_moments.py b/tests/test_integration_rule_moments.py
index ba3f1347..9f7ce1f3 100644
--- a/tests/test_integration_rule_moments.py
+++ b/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:
diff --git a/tests/test_moments.py b/tests/test_moments.py
index 283eb266..c4eab398 100644
--- a/tests/test_moments.py
+++ b/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():
diff --git a/tests/test_rule_moment_judgments.py b/tests/test_rule_moment_judgments.py
new file mode 100644
index 00000000..611fb051
--- /dev/null
+++ b/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
diff --git a/tests/test_rule_moment_signal_hooks.py b/tests/test_rule_moment_signal_hooks.py
new file mode 100644
index 00000000..be703a05
--- /dev/null
+++ b/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)
diff --git a/tests/test_services_backup.py b/tests/test_services_backup.py
index 7ed9b45b..95defdee 100644
--- a/tests/test_services_backup.py
+++ b/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] == []