FabledGrimoire/src/grimoire/services/rule_usage.py

"""Rule usage telemetry — did a surfaced rule ever get read?

The sibling of `note_usage`, for rules.

Event streams, deliberately independent:

  - SURFACED: a surface put this rule in front of the agent — a ranked arm,
    unbidden, or a bulk delivery (see AMBIENT VS RANKED below).
  - PULLED: someone then opened it in full (`get_rule`, or the REST detail
    route).

WHY RULES NEED THEIR OWN STREAM. `retrieval_logs` records what the ranker
scored, never whether the hint was any use, so even an honest clear-rate cannot
settle whether a rule arm earns its lines. The ratio these two streams produce
is the missing half, and without it any threshold change is a number picked
off a histogram. (A clear-rate can also mislead outright: an arm that logs
only the calls that FOUND something reports a decline count of zero at any
threshold — a statistic that cannot vary, read as a finding about the corpus.)

Design notes, mirroring `note_usage`:
  - Writes are fire-and-forget through `background.spawn`, so telemetry never
    adds latency to — or can break — the surface it observes. This module does
    NOT carry its own copy of the strong-reference dance; `background` is the
    one place that gets it right, and another copy is how one of them drifts.
  - Failures degrade, but never SILENTLY. `report_telemetry_failure` logs at
    WARNING and drops one AppLog row per process per site. A telemetry
    subsystem whose failures go to `logger.debug` can run at zero for weeks,
    indistinguishable from "nobody uses this".
  - Reads (`usage_for_rules`) are awaited and aggregated in one round-trip for
    a whole page, never per row.

AMBIENT VS RANKED. The note twin splits ranked surfacings from ambient ones
because `enter_project` and the skill sync put records in front of the agent
without choosing them, and counting those as surfacings makes recency read as
popularity. Rules have exactly that shape: the SessionStart preload, and every
`rules_payload` surface hand over the whole applicable set at once, chosen by
nobody.

Those bulk surfaces emit too, as ambient. Without them the always-on set's
token cost would be certain and its usefulness UNFALSIFIABLE — the one surface
whose value is actually in question would be the one exempt from the
scoreboard that judges every other. The split is a readout-level `case()`, no
migration, because `event` and `source` are plain Text with no CHECK
constraint. `source` stays granular so a reader can still tell the preload
from `enter_project` from a ranked arm.

WHY THIS NAMES THE RANKED SOURCES AND THE TWIN NAMES THE AMBIENT ONES. A
deliberate divergence, on the failure mode rather than on symmetry. Both shapes
fail silently when someone adds a surface and forgets the list, so the question
is which list changes more often — and here it is the ambient one. Ranked
sources are added when somebody builds a ranker, which is rare and deliberate;
bulk ones appear whenever a surface hands rules over, which is most of them.
Naming the rare, slow-moving half means a newly-added bulk surface defaults to
`ambient`, which merely under-counts it, instead of defaulting to `ranked`,
which would quietly pad the pull-through denominator with surfacings nobody
chose and make the arm look imprecise. Keep the list that must be remembered
as short and as slow-moving as possible.
"""
from __future__ import annotations

import logging
from datetime import datetime

from sqlalchemy import case, func, select, true

from grimoire.models import async_session
from grimoire.models.base import iso
from grimoire.models.rule_usage import (
    APPLIED, DEPARTED, OUTCOMES, PULLED, SURFACED, RuleUsageEvent,
)
from grimoire.services.background import report_telemetry_failure, spawn
from grimoire.services.retrieval_pipeline import MOMENT_RULE_SOURCE, RULE_RANKED_SOURCES

logger = logging.getLogger(__name__)

# The surfaces that CHOSE the rules they showed. Everything else is ambient —
# see the module docstring for why the rare half is the half that gets named.
#
# Membership is the whole definition of the pull-through denominator: a ranked
# surfacing is a claim ("this rule may apply to what you are doing") that a pull
# can confirm or refute, while an ambient one is a delivery nobody decided on.
# The ranked half is read off the retrieval pipeline's specs: every rule arm
# — the reply backstop included, which records only
# the rule that HELD the reply — the preference slot, which is a ranker's
# choice twice over, and the rule reached through a confirmed lesson link,
# whose own source lets its pull-through read apart from a direct match.
# An arm added there is counted here by construction.
RANKED_SOURCES = (
    *RULE_RANKED_SOURCES,
    # A rule mounted on a moment. Not a ranker's pick, but
    # not bulk either: somebody decided this rule applies at this moment, and
    # that is exactly the claim a pull can confirm or refute. Its
    # pull-through is the evidence for whether a mount earns its line.
    MOMENT_RULE_SOURCE,
)


def is_ambient(source: str) -> bool:
    """Was this surfacing a bulk delivery rather than a ranked choice?

    One definition, read by both the per-rule badge readout and the aggregate
    in `retrieval_telemetry`, so the two cannot disagree about what
    "surfaced" counts.

    Sync and pure, per the service canon, but deliberately PUBLIC where
    that canon says such helpers stay `_private`. The departure is the point:
    a module-private copy in each caller is exactly the second definition this
    exists to prevent.
    """
    return source not in RANKED_SOURCES


async def _report_failure(site: str) -> None:
    await report_telemetry_failure("rule_usage", site)


async def _insert_events(rows: list[dict]) -> None:
    """Persist usage rows. Best-effort: failures degrade, visibly."""
    try:
        async with async_session() as session:
            session.add_all([RuleUsageEvent(**row) for row in rows])
            await session.commit()
    except Exception:
        await _report_failure("write")


def _schedule(rows: list[dict]) -> None:
    if not rows:
        return
    spawn(_insert_events(rows), site="rule_usage_write")


def record_rule_surfaced(
    *, user_id: int | None, rule_ids: list[int] | set[int], source: str,
    detail: dict[int, str] | None = None,
) -> None:
    """Fire-and-forget: record that these rules were shown to the agent.

    Takes the whole delivery at once — one insert per surfacing event, not per
    rule — because a hint is a single decision and its rows should land
    together.

    Record what was actually SHOWN, never what was considered. For the ranked
    arm that means the post-filter hits: it drops what the session already
    holds (`exclude_rule_ids`) before it speaks, and a rule considered and not
    shown was not surfaced. Counting those would inflate the denominator with
    claims the agent never saw, which reads as a precision problem the arm does
    not have.

    Bulk surfaces pass their whole delivered set, which is the same rule read
    from the other end — everything in a preload IS shown. `source` is what
    separates the two afterwards (see `RANKED_SOURCES`); this function does not
    care which kind it is recording.

    `detail` maps a rule id to what to record beside its row — the moment a
    mounted rule arrived at, so a mount's pull-through can be
    read per moment rather than only per arm.
    """
    try:
        rows = [
            {
                "user_id": user_id,
                "rule_id": int(rid),
                "event": SURFACED,
                "source": source,
                **({"detail": detail[rid]} if detail and rid in detail else {}),
            }
            for rid in rule_ids
        ]
    except Exception:
        logger.debug("rule usage payload build failed", exc_info=True)
        return
    _schedule(rows)


def record_rule_pulled(*, user_id: int | None, rule_id: int, source: str) -> None:
    """Fire-and-forget: record that a rule was opened in full.

    A PULL is somebody choosing to open one record. `enter_project` is NOT a
    pull — they are bulk resident loads that hand over
    every applicable rule at once, and counting them would swamp the signal
    with the very ambient delivery the ratio exists to distinguish from.
    """
    try:
        rows = [
            {
                "user_id": user_id,
                "rule_id": int(rule_id),
                "event": PULLED,
                "source": source,
            }
        ]
    except Exception:
        logger.debug("rule usage payload build failed", exc_info=True)
        return
    _schedule(rows)


def record_rule_outcome(
    *,
    user_id: int | None,
    rule_id: int,
    outcome: str,
    source: str,
    detail: str = "",
) -> None:
    """Fire-and-forget: record what a rule ACTUALLY CHANGED.

    The third stream. `surfaced` says the system offered a rule; `pulled`
    says somebody opened it. Neither says whether it made any difference, so
    without this a rule that fires constantly and is always obeyed and a rule
    that fires constantly and is never obeyed produce identical telemetry.
    The second is far the more urgent.

    Two outcomes, because there are only two a judge can honestly report:

      APPLIED  — the rule changed what was done, or confirmed it. No `detail`
                 required: following a rule is the unremarkable case and
                 charging prose for it is how an event stops being recorded.
      DEPARTED — read, and deliberately not followed. `detail` is REQUIRED
                 and is the entire value of the event. A departure without
                 its reason reads back as a miss, which collapses the two
                 states this exists to separate.

    THERE IS NO THIRD CALL, and the absence is the design. Read-and-silently-
    unchanged is real — it is the failure this stream exists to expose — but
    it cannot be reported, because an agent that knew it was ignoring a rule
    would not be ignoring it. It is derived: a rule pulled, with no outcome
    behind it. See `outcome_state`.

    Guarded rather than trusting: a bad outcome or a reasonless departure is
    dropped and REPORTED, never written. Telemetry that lies is worse than
    telemetry that is missing, and a `departed` row with an empty
    reason is a lie the readout cannot detect.
    """
    if outcome not in OUTCOMES:
        logger.warning("rule outcome rejected: unknown outcome %r", outcome)
        spawn(_report_failure("outcome_unknown"), site="rule_usage_outcome")
        return
    if outcome == DEPARTED and not (detail or "").strip():
        logger.warning("rule outcome rejected: departure with no reason")
        spawn(_report_failure("outcome_no_reason"), site="rule_usage_outcome")
        return
    try:
        rows = [
            {
                "user_id": user_id,
                "rule_id": int(rule_id),
                "event": outcome,
                "source": source,
                "detail": (detail or "").strip() or None,
            }
        ]
    except Exception:
        logger.debug("rule usage payload build failed", exc_info=True)
        return
    _schedule(rows)


# What a rule's usage says happened to it, in one word. The four states are
# ordered by how much the system actually knows; recorded outcomes are what
# keep UNACTED from hiding inside FOLLOWED.
UNREAD = "unread"        # surfaced, never opened
UNACTED = "unacted"      # opened, and nothing recorded after — the blind spot
FOLLOWED = "followed"    # opened and applied
DEPARTED_FROM = "departed"   # opened and deliberately not followed, with a why


def outcome_state(usage: dict) -> str:
    """Which of the four states a rule's usage is in.

    Pure, and reading only the aggregate `usage_for_rules` already returns —
    so the readout, the badge and any later session summary all answer this
    question the same way. Two callers computing "was this followed" from raw
    counts would drift.

    PRECEDENCE, and it is deliberate: a departure outranks an application.
    A rule both applied and departed from in the same window is a rule
    someone argued with, and the argument is the interesting half — reporting
    it as plain compliance would hide the one row a reader most wants.

    UNACTED is the derived state and the reason this function exists. It is
    not "no data"; it is a rule that was surfaced, deliberately OPENED, and
    then left no trace of having mattered. That is a much stronger signal
    than never having been opened at all, and without recorded outcomes it
    is indistinguishable from compliance.
    """
    if int(usage.get("departed_count") or 0):
        return DEPARTED_FROM
    if int(usage.get("applied_count") or 0):
        return FOLLOWED
    if int(usage.get("pull_count") or 0):
        return UNACTED
    return UNREAD


def empty_rule_usage() -> dict:
    """The zero readout — what a rule with no recorded events looks like.

    Callers render this shape unconditionally, so a rule predating the table
    reads as "never surfaced, never pulled" rather than as a missing key. That
    distinction matters more here than for notes: every rule in an install
    predates this table, so for a while "no events" is the normal state and it
    must not look like a broken readout.

    `surfaced_count` is RANKED surfacings only; `ambient_count` is the bulk
    deliveries (see `RANKED_SOURCES`). The split is what keeps the badge's
    "shown often, opened never → dead weight" reading honest: every rule in an
    always-on set is delivered every session, so an unsplit counter would rank
    the resident set as the most-surfaced rules in the install purely for being
    resident.
    """
    return {
        "surfaced_count": 0,
        "ambient_count": 0,
        "pull_count": 0,
        # The outcome half. Zero here means "nothing recorded", which for a
        # rule that was also never pulled is simply silence — and for one
        # that WAS pulled is the blind spot (UNACTED).
        # `outcome_state` is what tells those apart; no caller should be
        # reading these counts raw to decide it.
        #
        # The REASON for a departure is on the row (`detail`), not here. One
        # GROUP BY cannot carry the text of the latest departure without a
        # DISTINCT ON alongside it, and a key that the aggregate could never
        # fill would read as "no reason given" on every rule that has one —
        # a permanently-null field that lies. The readout that needs the
        # prose reads the rows.
        "applied_count": 0,
        "departed_count": 0,
        "last_outcome_at": None,
        "last_surfaced_at": None,
        "last_pulled_at": None,
    }


async def usage_for_rules(
    rule_ids: list[int], since: datetime | None = None,
) -> dict[int, dict]:
    """Aggregate usage for a set of rules: {rule_id: {counts + timestamps}}.

    One GROUP BY for the whole page rather than a query per row — this feeds a
    list view, so the per-row shape would be N+1 by construction. Rules with no
    events come back with `empty_rule_usage()`, so the caller never has to tell
    "no events" from "not in the result".

    `since` limits the counts to events at or after it, so two windows can be
    compared; without it they cover everything recorded.
    """
    ids = [int(r) for r in rule_ids]
    out: dict[int, dict] = {rid: empty_rule_usage() for rid in ids}
    if not ids:
        return out

    # Classified in SQL so the group stays small: per rule we get at most
    # (surfaced-ranked, surfaced-ambient, pulled) rather than a row per distinct
    # source. ONE labelled expression, bound to a variable and reused in the
    # GROUP BY — a second `case()` instance there renders its own expanding-IN
    # bind names under asyncpg, so the database sees two DIFFERENT expressions
    # and rejects the query with a GroupingError. The note twin carries the
    # same warning for the same reason: the rejection is swallowed and every
    # counter reads zero while the writes are landing fine.
    ambient = case(
        (RuleUsageEvent.source.notin_(RANKED_SOURCES), True),
        else_=False,
    ).label("ambient")
    try:
        async with async_session() as session:
            rows = (
                await session.execute(
                    select(
                        RuleUsageEvent.rule_id,
                        RuleUsageEvent.event,
                        func.count().label("n"),
                        func.max(RuleUsageEvent.created_at).label("last_at"),
                        ambient,
                    )
                    .where(RuleUsageEvent.rule_id.in_(ids))
                    .where(
                        RuleUsageEvent.created_at >= since
                        if since is not None else true()
                    )
                    .group_by(
                        RuleUsageEvent.rule_id,
                        RuleUsageEvent.event,
                        ambient,
                    )
                )
            ).all()
    except Exception:
        # A telemetry readout must not be able to break the list it decorates —
        # but it must say it failed, or a broken readout is indistinguishable
        # from a corpus nobody uses.
        await _report_failure("readout")
        return out

    for rule_id, event, n, last_at, is_amb in rows:
        slot = out.get(int(rule_id))
        if slot is None:
            continue
        # Outcomes first, and never split by ambient. `ambient` asks whether
        # a RANKER chose to show the rule; an outcome is reported by a judge
        # after the fact and has no ranker behind it, so the flag is noise
        # here. Branching on it would silently drop every outcome row into a
        # bucket nothing reads.
        if event in OUTCOMES:
            key = "applied_count" if event == APPLIED else "departed_count"
            slot[key] = slot[key] + int(n)
            prev = slot["last_outcome_at"]
            now = iso(last_at)
            if now and (prev is None or now > prev):
                slot["last_outcome_at"] = now
        elif event == SURFACED and is_amb:
            slot["ambient_count"] = int(n)
        elif event == SURFACED:
            slot["surfaced_count"] = int(n)
            slot["last_surfaced_at"] = iso(last_at)
        elif event == PULLED:
            # Pulls are pulls regardless of what surfaced the rule — "did
            # anyone ever open this?" does not depend on how it was found. Both
            # halves accumulate, so this ADDS rather than assigns: a rule can
            # now be pulled after a ranked hint and after a preload, and the
            # split arrives as two rows.
            slot["pull_count"] = slot["pull_count"] + int(n)
            latest = iso(last_at)
            if latest and (slot["last_pulled_at"] or "") < latest:
                slot["last_pulled_at"] = latest
    return out