feat(moments): a mount that keeps arriving where it does not apply proposes its own removal (milestone 458 step 7b, #4955)

The open-after-moment signal proposes a mount; nothing proposed taking one
off, so a wrong mount was noise at every occurrence until someone happened
to notice. rule_misfired(rule_id, moment, why, reached_by) records a report
against a MOUNTED pair, counted per distinct day (the MCP door carries no
session id) on a new rule_moment_judgments.misfire column (migration 0119,
backup v24). At three days the response carries a line asking the agent to
offer the operator the fix - reject takes the rule off, unmap_action stops
the action reaching the moment, confirm keeps the mount and stops the
asking - and Settings > Moments lists it as an unmount proposal with the
reasons and the actions that reached it. A re-mount clears the count.

Taught in moments.md, missed-retrieval.md and the reply hold's wording.

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

tests/test_integration_rule_moment_judgments.py

@@ -7,7 +7,10 @@ What the step promises, against the real tables:
   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.
+- only the owner can propose, judge, or feed the signal;
+- the misfire (step 7b): only a mount can misfire, reports count DAYS, the
+  bar makes an unmount proposal, reject takes it off, confirm keeps it and
+  stops the asking, and a re-mount starts the count again.
 """
 import pytest
 import pytest_asyncio
@@ -191,3 +194,139 @@ async def test_the_open_resolves_acts_through_the_installs_mappings(world):
     assert "work.deliver" in out["moments"]
     assert "work.run" not in out["moments"]
     assert (await _row(done.id, "work.deliver")).source == "signal"
+
+
+# ── The misfire (step 7b) ────────────────────────────────────────────────────
+
+
+@pytest.fixture
+def days(monkeypatch):
+    """Each report lands on the day named next, so a test can cross days
+    without waiting for them. `misfire_situation` is the one place a day
+    becomes a situation, so this is what a real date change does."""
+    queue: list[str] = []
+
+    def situation(_now):
+        from scribe.services import lesson_rules
+        return lesson_rules.situation_key("d", queue.pop(0))
+
+    monkeypatch.setattr(judgments_svc, "misfire_situation", situation)
+    return queue
+
+
+async def _misfire(uid, rule_id, moment="work.finish", why="closing a docs-only task",
+                   reached_by="status=done"):
+    return await judgments_svc.misfired(uid, rule_id, moment, why=why, reached_by=reached_by)
+
+
+async def test_only_a_mount_can_misfire_and_only_with_a_reason(world, days):
+    uid, sid, done = world["uid"], world["sid"], world["done"]
+    out = await _misfire(uid, done.id)
+    assert not out["recorded"] and "not mounted on work.finish" in out["error"]
+    assert "when_to_apply" in out["error"]
+    await rulebooks_svc.set_rule_moments(done.id, uid, ["work.finish"])
+    out = await judgments_svc.misfired(uid, done.id, "work.finish", why="  ")
+    assert not out["recorded"] and "why" in out["error"]
+    out = await judgments_svc.misfired(uid, done.id, "work.finished", why="x")
+    assert not out["recorded"] and "work.finished" in out["error"]
+    out = await _misfire(sid, done.id)
+    assert out == {"recorded": False, "error": "not a rule you own"}
+    assert (await _row(done.id, "work.finish")).misfire is None
+
+
+async def test_misfires_count_days_and_propose_the_unmount(world, days):
+    uid, done = world["uid"], world["done"]
+    await rulebooks_svc.set_rule_moments(done.id, uid, ["work.finish"])
+    days.extend(["2026-10-01", "2026-10-01", "2026-10-02"])
+    for _ in range(3):
+        out = await _misfire(uid, done.id)
+        assert out["recorded"] and out["context"] == ""
+    assert out["days"] == 2 and out["bar"] == 3
+    assert (await judgments_svc.pending(uid, rule_id=done.id))["rules"] == []
+
+    days.append("2026-10-03")
+    out = await _misfire(uid, done.id, why="closing a planning stub",
+                         reached_by="status=cancelled")
+    line = out["context"]
+    assert out["days"] == 3
+    assert f"#{done.id}" in line and "`work.finish`" in line and "3 distinct days" in line
+    assert "`status=done` ×3" in line and "`status=cancelled` ×1" in line
+    assert '"verdict": "reject"' in line and "unmap_action" in line and '"confirm"' in line
+    # Asked once; the cooldown keeps the next report quiet.
+    days.append("2026-10-04")
+    assert (await _misfire(uid, done.id))["context"] == ""
+
+    [entry] = (await judgments_svc.pending(uid, rule_id=done.id))["rules"]
+    assert entry["mounted"] == ["work.finish"]
+    [proposal] = entry["proposals"]
+    assert (proposal["proposal"], proposal["source"], proposal["moment"]) == (
+        "unmount", "misfire", "work.finish")
+    assert proposal["evidence"]["situations"] == 4
+    assert proposal["reached_by"] == {"status=done": 4, "status=cancelled": 1}
+    assert len(proposal["reasons"]) == judgments_svc.MISFIRE_REASONS
+    assert proposal["why"] == "closing a docs-only task"
+
+
+async def _past_the_bar(uid, rule_id, days, moment="work.finish"):
+    days.extend(["2026-10-01", "2026-10-02", "2026-10-03"])
+    for _ in range(3):
+        out = await _misfire(uid, rule_id, moment=moment)
+    assert out["context"]
+
+
+async def test_reject_takes_the_mount_off(world, days):
+    uid, done = world["uid"], world["done"]
+    await rulebooks_svc.set_rule_moments(done.id, uid, ["work.finish", "reply.report"])
+    await _past_the_bar(uid, done.id, days)
+    out = await judgments_svc.judge(uid, [
+        {"rule_id": done.id, "moment": "work.finish", "verdict": "reject", "note": "not at task close"},
+    ])
+    assert out["refused"] == []
+    assert (await rulebooks_svc.list_rule_moments([done.id]))[done.id] == ["reply.report"]
+    assert (await _row(done.id, "work.finish")).state == "rejected"
+    assert (await judgments_svc.pending(uid, rule_id=done.id))["rules"] == []
+    # And it cannot misfire where it no longer is.
+    assert not (await _misfire(uid, done.id))["recorded"]
+
+
+async def test_confirm_keeps_the_mount_and_stops_the_asking(world, days):
+    uid, done = world["uid"], world["done"]
+    await rulebooks_svc.set_rule_moments(done.id, uid, ["work.finish"])
+    await _past_the_bar(uid, done.id, days)
+    await judgments_svc.judge(uid, [
+        {"rule_id": done.id, "moment": "work.finish", "verdict": "confirm", "note": "it does belong"},
+    ])
+    assert (await rulebooks_svc.list_rule_moments([done.id]))[done.id] == ["work.finish"]
+    row = await _row(done.id, "work.finish")
+    assert row.state == "confirmed" and row.misfire["kept_note"] == "it does belong"
+    assert (await judgments_svc.pending(uid, rule_id=done.id))["rules"] == []
+    # Reports still count, and ask nothing — past the cooldown too.
+    days.extend(["2026-10-10", "2026-10-11"])
+    for _ in range(2):
+        out = await _misfire(uid, done.id)
+        assert out["recorded"] and out["context"] == ""
+    assert out["kept"]["note"] == "it does belong"
+
+
+async def test_a_remount_starts_the_count_again(world, days):
+    uid, done = world["uid"], world["done"]
+    await rulebooks_svc.set_rule_moments(done.id, uid, ["work.finish"])
+    await _past_the_bar(uid, done.id, days)
+    await rulebooks_svc.set_rule_moments(done.id, uid, [])
+    await rulebooks_svc.set_rule_moments(done.id, uid, ["work.finish"])
+    row = await _row(done.id, "work.finish")
+    assert row.state == "confirmed" and row.misfire is None
+
+
+async def test_a_mount_with_no_judgment_row_gets_one_on_its_first_misfire(world, days):
+    """Mounts made before migration 0118 have no row; a misfire must still count."""
+    uid, done = world["uid"], world["done"]
+    await rulebooks_svc.set_rule_moments(done.id, uid, ["work.finish"])
+    async with async_session() as s:
+        await s.delete(await s.get(RuleMomentJudgment, (await _row(done.id, "work.finish")).id))
+        await s.commit()
+    days.append("2026-10-01")
+    assert (await _misfire(uid, done.id))["recorded"]
+    row = await _row(done.id, "work.finish")
+    assert (row.state, row.source) == ("confirmed", "edit")
+    assert len(row.misfire["situations"]) == 1