"""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.

THE MISFIRE (step 7b) is the same machinery pointed the other way. A session
that was handed a mounted rule at a moment where it did not apply says so
(`misfired`), with why and the action that reached the moment. Reports count
per distinct DAY — the MCP door is stateless and no session id reaches it, and
a misfire seen on three days is a pattern where three in one afternoon may be
one piece of work. At the bar the pair becomes an UNMOUNT proposal; the
operator takes the rule off (reject), or unmaps the action, or keeps it
(confirm), which stops the asking. A rule left unopened is never counted:
most delivered rules go unopened, and the reply hold forces an open.

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"})

# The reasons a misfire proposal carries, newest kept. Enough for the operator
# to see whether the reports agree; few enough to read in the panel.
MISFIRE_REASONS = 5
# Distinct actions counted per pair. A moment is reached by a handful of
# actions; past this the counts stop telling the operator which one to unmap.
_MISFIRE_ACTIONS_CAP = 10
_MISFIRE_WHY_CHARS = 400
# On a mounted pair that has no judgment row — mounted before the rows were
# kept (migration 0118) — when its first misfire gives it one.
_UNRECORDED_MOUNT_NOTE = "mounted before judgments were recorded"


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({
            "proposal": "mount",
            "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,
        })
    for judgment, rule, mounted in await _misfire_proposals(user_id, rule_id, limit):
        entry = grouped.setdefault(rule.id, {
            **_brief(rule),
            "mounted": sorted(mounted, key=lambda n: (order.get(n, len(order)), n)),
            "proposals": [],
        })
        entry["proposals"].append(_unmount_item(judgment))
    rules = sorted(grouped.values(), key=lambda r: r["id"])
    return {"rules": rules, "total": sum(len(r["proposals"]) for r in rules)}


def _misfire_due_clause():
    """Proposed by the misfire signal and not yet kept by the operator."""
    mf = RuleMomentJudgment.misfire
    return mf["proposed_at"].astext.isnot(None), mf["kept_at"].astext.is_(None)


async def _misfire_proposals(user_id: int, rule_id: int | None, limit: int):
    """The unmount proposals: pairs past the misfire bar, unanswered, and
    still mounted — a pair taken off since has had its answer."""
    from scribe.services.rulebooks import _owned_rules_clause

    async with async_session() as session:
        stmt = (
            select(RuleMomentJudgment, Rule)
            .join(Rule, Rule.id == RuleMomentJudgment.rule_id)
            .where(_owned_rules_clause(user_id), *_misfire_due_clause())
        )
        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})
    return [(j, r, mounts.get(r.id, set())) for j, r in rows
            if j.moment in mounts.get(r.id, set())]


def _unmount_item(judgment: RuleMomentJudgment) -> dict:
    mf = judgment.misfire or {}
    reasons = list(mf.get("reasons") or [])
    return {
        "proposal": "unmount",
        "moment": judgment.moment,
        "source": "misfire",
        "why": reasons[-1]["why"] if reasons else "",
        "reasons": reasons,
        "reached_by": dict(mf.get("reached_by") or {}),
        "evidence": lesson_rules.evidence_summary(mf),
        "created_at": mf.get("first_at"),
    }


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) —
    on a pair already mounted, it KEEPS the mount and answers a misfire
    proposal; 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)
        # A fresh mount is a fresh judgment: misfires counted against an
        # earlier mount of the pair are not evidence against this one.
        rows[(rule_id, moment)].misfire = None
    for moment in removed:
        put(moment, REJECTED, _UNMOUNTED_NOTE)
    for moment in explicit:
        if verdict in (CONFIRMED, REJECTED):
            put(moment, verdict, "")
        if verdict == CONFIRMED and moment in after:
            _keep(rows[(rule_id, moment)], note, now)
    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


def _keep(row: RuleMomentJudgment, note: str, now: datetime) -> None:
    """The operator kept a mount the misfire signal proposed taking off. The
    reports stay; the asking stops."""
    mf = row.misfire
    if not isinstance(mf, dict) or not mf.get("proposed_at") or mf.get("kept_at"):
        return
    row.misfire = {**mf, "kept_at": now.isoformat(), "kept_note": note or ""}


# ── 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}


# ── The misfire: a mounted rule that arrived where it does not apply ─────────


def misfire_situation(now: datetime) -> str:
    """One situation per UTC day. The MCP door carries no session id; a day
    is the coarser, honest unit, and the slower of the two to cross the bar."""
    return lesson_rules.situation_key("d", now.date().isoformat())


def add_misfire(misfire, key: str, project_id: int | None, now: datetime, *,
                why: str, reached_by: str) -> dict:
    """A NEW misfire dict with this report counted. Pure, like
    `lesson_rules.add_evidence`, whose counting it reuses: the per-day
    situations and the bar are the same as every other proposal's."""
    ev = lesson_rules.add_evidence(misfire, key, project_id, now)
    reasons = list(ev.get("reasons") or [])
    reasons.append({"why": why[:_MISFIRE_WHY_CHARS], "reached_by": reached_by,
                    "at": now.isoformat()})
    ev["reasons"] = reasons[-MISFIRE_REASONS:]
    actions = dict(ev.get("reached_by") or {})
    if reached_by and (reached_by in actions or len(actions) < _MISFIRE_ACTIONS_CAP):
        actions[reached_by] = int(actions.get(reached_by) or 0) + 1
    ev["reached_by"] = actions
    return ev


def _unmount_line(rule: Rule, moment: str, misfire: dict) -> str:
    days = len(misfire.get("situations") or [])
    actions = ", ".join(f"`{a}` ×{n}" for a, n in sorted(
        (misfire.get("reached_by") or {}).items(), key=lambda kv: -kv[1]))
    kind = rule.kind or "rule"
    return (
        f"> {kind.capitalize()} #{rule.id} \"{rule.title}\" has been reported arriving at "
        f"`{moment}` where it did not apply, on {days} distinct days"
        + (f" (reached by {actions})" if actions else "")
        + ". Offer the operator the fix in one line. If the rule does not belong "
        f"at that moment, `judge_rule_moments([{{\"rule_id\": {rule.id}, \"moment\": "
        f"\"{moment}\", \"verdict\": \"reject\", \"note\": \"why\"}}])` takes it off. "
        "If it is the ACTION that is not that moment here, `unmap_action` for "
        "it instead (`list_moments` shows the mapping). If they keep it, "
        "`\"confirm\"` with their why stops the asking."
    )


async def misfired(
    user_id: int, rule_id: int, moment: str, *, why: str,
    reached_by: str = "", project_id: int | None = None,
) -> dict:
    """Record that a mounted rule arrived at `moment` where it did not apply.

    Only a MOUNT can misfire: a rule that arrived by resemblance is a
    retrieval question, answered by its trigger. `why` is required — a report
    without its reason cannot tell the operator what to fix. Returns the
    count so far and, once the pair crosses the bar, `context`: the line
    asking the reader to offer the unmount. A kept mount still counts its
    reports and asks nothing.
    """
    try:
        name = moments_svc.require_moment((moment or "").strip())
    except ValueError as exc:
        return {"recorded": False, "error": str(exc)}
    why = (why or "").strip() if isinstance(why, str) else ""
    if not why:
        return {"recorded": False, "error": (
            "say why it did not apply here — the reason is what tells the "
            "operator whether the rule or the action is wrong")}
    rid = _int(rule_id)
    reached_by = (reached_by or "").strip()[:200] if isinstance(reached_by, str) else ""
    now = datetime.now(timezone.utc)
    async with async_session() as session:
        owned = await _owned_rules(session, user_id, [rid] if rid else [])
        rule = owned.get(rid) if rid else None
        if rule is None:
            return {"recorded": False, "error": "not a rule you own"}
        mounted = (await _mounts(session, [rid])).get(rid, set())
        if name not in mounted:
            return {"recorded": False, "error": (
                f"rule {rid} is not mounted on {name}"
                + (f" (it is on {', '.join(sorted(mounted))})" if mounted else "")
                + ". A rule that arrived by its words rather than a mount is "
                "fixed through its trigger — update_rule(when_to_apply=...).")}
        row = (await _rows(session, [rid])).get((rid, name))
        if row is None:
            row = RuleMomentJudgment(rule_id=rid, moment=name, state=CONFIRMED,
                                     source="edit", note=_UNRECORDED_MOUNT_NOTE,
                                     judged_at=now)
            session.add(row)
        mf = add_misfire(row.misfire, misfire_situation(now), project_id, now,
                         why=why, reached_by=reached_by)
        kept = bool(mf.get("kept_at"))
        due = not kept and lesson_rules.proposal_due(mf, now)
        if due:
            mf["proposed_at"] = now.isoformat()
            mf["proposed_count"] = int(mf.get("proposed_count") or 0) + 1
        row.misfire = mf
        try:
            await session.commit()
        except IntegrityError:
            await session.rollback()
            return {"recorded": False, "error": "a report for this pair landed at the same moment; try again"}
    out = {
        "recorded": True, "rule_id": rid, "moment": name,
        "days": len(mf.get("situations") or []),
        "bar": lesson_rules.PROPOSE_SITUATIONS,
        "context": _unmount_line(rule, name, mf) if due else "",
    }
    if kept:
        out["kept"] = {"at": mf.get("kept_at"), "note": mf.get("kept_note") or ""}
    return out