fix(6470): a moment proposal counts only opens no hook explained, and asks the reader to judge it

The signal counted every get_rule open against every moment from the
previous three minutes, so a rule a search arm had just shown was
proposed for whatever moment happened to fire. All nine pending signal
proposals were wrong.

- The hook skips an open of a rule a hook named this session, and sends
  the last three acts from 90 seconds instead of 30 from 180.
- A rule already mounted anywhere gathers no signal evidence.
- A signal pair keeps the actions that reached its moment and shows them
  as its why.
- The proposal line says it is co-occurrence, not a recommendation, and
  asks the reader to check the rule against the moment before offering
  it; moments.md says the same.

Plugin 2026.10.26.2029.

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

src/grimoire/services/rule_moment_judgments.py

@@ -494,17 +494,50 @@ def _keep(row: RuleMomentJudgment, note: str, now: datetime) -> None:
 # ── The signal ───────────────────────────────────────────────────────────────
 
 
+# How many distinct actions a signal pair remembers as having reached its
+# moment. Enough to show the operator what the opens followed.
+_SIGNAL_ACTIONS_CAP = 5
+
+
+def _reached_phrase(evidence: dict) -> str:
+    reached = (evidence or {}).get("reached_by") or {}
+    if not isinstance(reached, dict) or not reached:
+        return ""
+    top = sorted(reached.items(), key=lambda kv: (-int(kv[1] or 0), kv[0]))[:3]
+    return " (reached by " + ", ".join(f"`{match}`" for match, _n in top) + ")"
+
+
+def _signal_note(moment: str, evidence: dict) -> str:
+    """The `why` a signal proposal shows the operator: what the opens followed."""
+    counts = lesson_rules.evidence_summary(evidence)
+    return (
+        f"Opened just after `{moment}`{_reached_phrase(evidence)} in "
+        f"{counts['situations']} distinct sessions, each time without a hook having "
+        f"shown the rule first. Co-occurrence only: confirm it if the rule governs "
+        f"that point in the work whatever the work is about."
+    )
+
+
 def _proposal_line(rule: Rule, moment: str, evidence: dict) -> str:
+    """The line an open past the bar returns to the session.
+
+    It says what the evidence is, and makes offering it conditional on the
+    reader's own reading: a session relays an unqualified "offer this" as a
+    recommendation, and co-occurrence is not one.
+    """
     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."
+        f"> {kind.capitalize()} #{rule.id} \"{rule.title}\" was opened just after "
+        f"`{moment}`{_reached_phrase(evidence)} in {counts['situations']} distinct "
+        f"sessions, each time without a hook having shown it first. That is "
+        f"co-occurrence, not a recommendation. Read the {kind}'s trigger against what "
+        f"`{moment}` means (`list_moments`): only if the {kind} governs that point in "
+        f"the work whatever the work is about, ask the operator whether to mount it, "
+        f"with your reason; if it is about a subject, leave it. On their yes, "
+        f"`judge_rule_moments([{{\"rule_id\": {rule.id}, \"moment\": \"{moment}\", "
+        f"\"verdict\": \"confirm\", \"note\": \"why\"}}])` mounts it; on a no, "
+        f"`\"reject\"` with their why stops the asking."
     )
 
 
@@ -521,24 +554,39 @@ def signal_moments(moments) -> list[str]:
 
 async def co_occurred(
     user_id: int, rule_id: int, moments, *, situation: str,
-    project_id: int | None = None,
+    project_id: int | None = None, reached: dict | 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.
+    judged gathers nothing, and neither does a rule already mounted anywhere:
+    it arrives by lookup at its own moments, so an open of it is far likelier
+    to follow that delivery than to reveal a missing moment. `reached` maps a
+    moment to the action that reached it, kept as the proposal's evidence.
+    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)
+        return await _co_occurred(user_id, rule_id, moments, situation, project_id,
+                                  reached or {})
     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:
+def _add_reached(evidence: dict, match: str) -> dict:
+    """`evidence` with one more open counted against the action that reached
+    the moment, keeping at most _SIGNAL_ACTIONS_CAP distinct actions."""
+    reached = dict(evidence.get("reached_by") or {})
+    match = (match or "").strip()[:80]
+    if match and (match in reached or len(reached) < _SIGNAL_ACTIONS_CAP):
+        reached[match] = int(reached.get(match) or 0) + 1
+    return {**evidence, "reached_by": reached}
+
+
+async def _co_occurred(user_id, rule_id, moments, situation, project_id, reached) -> str:
     wanted = signal_moments(moments)
     key = lesson_rules.situation_key("s", situation)
     rid = _int(rule_id)
@@ -551,11 +599,10 @@ async def _co_occurred(user_id, rule_id, moments, situation, project_id) -> str:
         rule = owned.get(rid)
         if rule is None:
             return ""
-        mounted = (await _mounts(session, [rid])).get(rid, set())
+        if (await _mounts(session, [rid])).get(rid):
+            return ""
         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
@@ -563,11 +610,13 @@ async def _co_occurred(user_id, rule_id, moments, situation, project_id) -> str:
                 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)
+            ev = _add_reached(ev, reached.get(moment, ""))
             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
+            row.note = _signal_note(moment, ev)
         try:
             await session.commit()
         except IntegrityError:
@@ -599,6 +648,7 @@ async def opened_after(
     from grimoire.services.moment_actions import moments_for
 
     reached: list[str] = []
+    reached_by: dict[str, str] = {}
     for act in list(acts or [])[:_ACTS_CAP]:
         if not isinstance(act, dict):
             continue
@@ -615,11 +665,13 @@ async def opened_after(
             name = hit.get("moment") if isinstance(hit, dict) else None
             if name and name not in reached:
                 reached.append(name)
+                reached_by[name] = str(hit.get("match") or tool)
     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,
+            reached=reached_by,
         )
     return {"moments": moments, "context": context}