FabledScribe/tests/test_services_rule_usage.py

"""Rule usage telemetry — the parts that need no database (milestone 333 step 1).

The round trip lives in `test_integration_backup_rule_usage_roundtrip.py`.
What is here is the payload building and the zero shape: cheap, and the half
where a mistake is silent rather than loud.
"""
import pytest

from scribe.models.rule_usage import (
    APPLIED, DEPARTED, PULLED, SURFACED, RuleUsageEvent,
)
from scribe.services import rule_usage


@pytest.fixture
def captured(monkeypatch):
    """Intercept the scheduler so the payload can be read without a loop.

    Patching `_schedule` rather than `background.spawn` keeps the test on this
    module's own seam: what is under test is which rows get built, not whether
    the shared fire-and-forget machinery works — that has its own home.
    """
    rows: list[list[dict]] = []
    monkeypatch.setattr(rule_usage, "_schedule", rows.append)
    return rows


def test_a_surfacing_records_one_row_per_rule(captured):
    """The arm shows a hint containing several rules at once; each needs its
    own row, because the readout is per rule."""
    rule_usage.record_rule_surfaced(
        user_id=7, rule_ids=[156, 157], source="write_path_rule"
    )
    [batch] = captured
    assert batch == [
        {"user_id": 7, "rule_id": 156, "event": SURFACED, "source": "write_path_rule"},
        {"user_id": 7, "rule_id": 157, "event": SURFACED, "source": "write_path_rule"},
    ]


def test_the_whole_hint_lands_as_one_batch(captured):
    """One scheduled insert for the hint, not one per rule. A hint is a single
    decision and its rows should land together — a partial batch would read as
    a hint that surfaced fewer rules than it did."""
    rule_usage.record_rule_surfaced(
        user_id=7, rule_ids=[1, 2, 3], source="write_path_rule"
    )
    assert len(captured) == 1
    assert len(captured[0]) == 3


def test_a_pull_records_one_row(captured):
    rule_usage.record_rule_pulled(user_id=7, rule_id=156, source="mcp_get_rule")
    assert captured == [
        [{"user_id": 7, "rule_id": 156, "event": PULLED, "source": "mcp_get_rule"}]
    ]


def test_an_actorless_event_is_still_recorded(captured):
    """The arm fires from a hook that may carry no authenticated user. Dropping
    those would silently shrink the denominator the ratio divides by — the
    surfacings would vanish while any later pull still counted."""
    rule_usage.record_rule_surfaced(
        user_id=None, rule_ids=[156], source="write_path_rule"
    )
    assert captured[0][0]["user_id"] is None


def test_an_empty_surfacing_builds_no_rows(captured):
    """The arm can rank everything out — `exclude_rule_ids` drops what the
    session already holds. That is not a surfacing, and the empty batch is
    where `_schedule` returns early rather than opening a session to insert
    nothing."""
    rule_usage.record_rule_surfaced(user_id=7, rule_ids=[], source="write_path_rule")
    assert captured == [[]]


def test_the_real_scheduler_returns_early_on_an_empty_batch():
    """The guard itself, against the REAL `_schedule` the stub above replaces.

    There is no running loop in a unit test, so `spawn` would be harmless
    anyway — but it would build a coroutine only to close it, and the point is
    that an empty batch never gets that far.
    """
    rule_usage._schedule([])  # must not raise


def test_a_bad_rule_id_is_dropped_not_raised(captured):
    """Telemetry must never break the surface it observes. An unconvertible id
    is a bug somewhere upstream, and the right response is to lose the row and
    log it — not to take down the write-path hint."""
    rule_usage.record_rule_pulled(
        user_id=7, rule_id="not-an-int", source="mcp_get_rule"  # type: ignore[arg-type]
    )
    assert captured == []


def test_the_zero_readout_names_every_key():
    """Callers render this shape unconditionally. Every rule in an existing
    install predates the table, so for a while "no events" is the NORMAL state
    — a missing key here would read as a broken readout on almost every row."""
    assert rule_usage.empty_rule_usage() == {
        "surfaced_count": 0,
        "ambient_count": 0,
        "pull_count": 0,
        "applied_count": 0,
        "departed_count": 0,
        "last_outcome_at": None,
        "last_surfaced_at": None,
        "last_pulled_at": None,
    }


def test_only_a_ranker_counts_as_ranked():
    """The bulk surfaces are ambient; the write-path arm is the only chooser.

    Inverted against the note twin on purpose (see the module docstring): the
    RARE half is the one that gets named, so a bulk surface added later and
    forgotten defaults to ambient — under-counting it — instead of defaulting
    to ranked and padding the pull-through denominator with surfacings nobody
    chose.
    """
    assert not rule_usage.is_ambient("write_path_rule")
    for bulk in (
        "session_start", "list_always_on_rules", "enter_project",
        "get_project", "get_milestone", "start_planning", "get_task",
    ):
        assert rule_usage.is_ambient(bulk), bulk
    # The safe default is the whole point of the inversion.
    assert rule_usage.is_ambient("some_surface_invented_next_year")


def test_the_model_serialises_the_fields_the_ratio_needs():
    ev = RuleUsageEvent(
        user_id=7, rule_id=156, event=SURFACED, source="write_path_rule"
    )
    row = ev.to_dict()
    assert row["rule_id"] == 156
    assert row["event"] == SURFACED
    assert row["source"] == "write_path_rule"
    # created_at is server-defaulted, so it is None until the row is flushed —
    # `iso()` must tolerate that rather than raising on a fresh instance.
    assert row["created_at"] is None


# ─── the readout (milestone 333 step 5) ──────────────────────────────────────
# Integration: a real GROUP BY over a real table. Step 1 unit-tested the WRITE
# path and the zero shape and left the aggregate uncovered, which only became
# load-bearing when the rule list started rendering it.


@pytest.mark.integration
@pytest.mark.asyncio
async def test_usage_for_rules_aggregates_per_rule(_dispose_engine):
    from sqlalchemy import delete

    from scribe.models import async_session
    from scribe.models.rule_usage import RuleUsageEvent

    async with async_session() as s:
        s.add_all([
            RuleUsageEvent(user_id=990020, rule_id=6001,
                           event=SURFACED, source="write_path_rule"),
            RuleUsageEvent(user_id=990020, rule_id=6001,
                           event=SURFACED, source="write_path_rule"),
            RuleUsageEvent(user_id=990020, rule_id=6001,
                           event=PULLED, source="mcp_get_rule"),
            RuleUsageEvent(user_id=990020, rule_id=6002,
                           event=SURFACED, source="write_path_rule"),
        ])
        await s.commit()
    try:
        out = await rule_usage.usage_for_rules([6001, 6002, 6003])

        assert out[6001]["surfaced_count"] == 2
        assert out[6001]["pull_count"] == 1
        assert out[6001]["last_surfaced_at"] is not None
        assert out[6001]["last_pulled_at"] is not None

        # Surfaced twice as often as it was opened — never, in this case.
        assert out[6002]["surfaced_count"] == 1
        assert out[6002]["pull_count"] == 0
        assert out[6002]["last_pulled_at"] is None

        # A rule with NO events still comes back, zero-filled. The caller must
        # never have to tell "no events" from "not in the result" — and on any
        # existing install that is nearly every rule.
        assert out[6003] == rule_usage.empty_rule_usage()

        # Nothing ambient in this fixture, so the ambient bucket stays empty
        # rather than absorbing the ranked hits.
        assert out[6001]["ambient_count"] == 0
    finally:
        async with async_session() as s:
            await s.execute(
                delete(RuleUsageEvent).where(RuleUsageEvent.user_id == 990020)
            )
            await s.commit()


@pytest.mark.integration
@pytest.mark.asyncio
async def test_usage_for_rules_on_an_empty_id_list_asks_the_database_nothing(
    _dispose_engine,
):
    """The list route calls this with whatever the page holds, which on an
    empty topic is nothing. An unguarded `IN ()` is both a pointless round trip
    and, on some drivers, a syntax error."""
    assert await rule_usage.usage_for_rules([]) == {}


@pytest.mark.integration
@pytest.mark.asyncio
async def test_a_preloaded_rule_does_not_read_as_a_ranked_surfacing(_dispose_engine):
    """The split that makes the always-on set judgeable (#3473).

    A resident rule is delivered every session by a surface that chose
    nothing. Counting those as `surfaced_count` would rank the always-on set
    as the most-surfaced rules in the install purely for being resident — and
    the badge's "shown often, opened never → dead weight" reading, which is
    the whole reason the counter exists, would then be exactly backwards.
    """
    from sqlalchemy import delete

    from scribe.models import async_session
    from scribe.models.rule_usage import RuleUsageEvent

    async with async_session() as s:
        s.add_all([
            # Delivered by the preload three times over: ambient, all of it.
            RuleUsageEvent(user_id=990021, rule_id=6101,
                           event=SURFACED, source="session_start"),
            RuleUsageEvent(user_id=990021, rule_id=6101,
                           event=SURFACED, source="list_always_on_rules"),
            RuleUsageEvent(user_id=990021, rule_id=6101,
                           event=SURFACED, source="enter_project"),
            # ...and once by the arm, which DID choose it.
            RuleUsageEvent(user_id=990021, rule_id=6101,
                           event=SURFACED, source="write_path_rule"),
            # Opened once after a hint and once from the list: pulls are pulls
            # however the rule was found, so both land in the one counter.
            RuleUsageEvent(user_id=990021, rule_id=6101,
                           event=PULLED, source="mcp_get_rule"),
            RuleUsageEvent(user_id=990021, rule_id=6101,
                           event=PULLED, source="rest_rule"),
        ])
        await s.commit()
    try:
        out = await rule_usage.usage_for_rules([6101])

        assert out[6101]["surfaced_count"] == 1, "only the arm chose this rule"
        assert out[6101]["ambient_count"] == 3, "three bulk deliveries"
        # Both PULLED rows accumulate — the loop ADDS rather than assigns, so a
        # rule opened after a hint and again from the list reports two, not one.
        assert out[6101]["pull_count"] == 2
        assert out[6101]["last_pulled_at"] is not None
    finally:
        async with async_session() as s:
            await s.execute(
                delete(RuleUsageEvent).where(RuleUsageEvent.user_id == 990021)
            )
            await s.commit()


# ── the outcome stream (#4212, milestone 419) ─────────────────────────────
#
# What these guard is a distinction, not a payload. Before this existed, a
# rule read and obeyed and a rule read and ignored left byte-identical
# telemetry, so the readout could not name the failure the whole milestone
# was opened on. The tests that matter most below are the ones asserting
# that a REASONLESS DEPARTURE IS NEVER WRITTEN, and that an unacted rule is
# a state in its own right rather than the absence of one.

def test_an_applied_outcome_needs_no_argument(captured):
    """Following a rule is the ordinary case. Charging prose for it would
    make the cheap event expensive, and an expensive event stops being
    recorded — which costs the whole measurement."""
    rule_usage.record_rule_outcome(
        user_id=7, rule_id=156, outcome=APPLIED, source="mcp_rule_outcome"
    )
    [batch] = captured
    assert batch == [{
        "user_id": 7, "rule_id": 156, "event": APPLIED,
        "source": "mcp_rule_outcome", "detail": None,
    }]


def test_a_departure_carries_its_reason(captured):
    rule_usage.record_rule_outcome(
        user_id=7, rule_id=156, outcome=DEPARTED, source="mcp_rule_outcome",
        detail="  the integration lane has no registry credentials  ",
    )
    [batch] = captured
    assert batch[0]["event"] == DEPARTED
    assert batch[0]["detail"] == "the integration lane has no registry credentials"


@pytest.mark.parametrize("reason", ["", "   ", "\n", None])
def test_a_departure_with_no_reason_is_never_written(captured, reason):
    """THE ONE THAT MATTERS. A `departed` row without its why reads back as a
    miss, so writing one would collapse the two states this table exists to
    separate — silently, in the readout, where nobody would see it happen.
    Dropped and reported, never stored: telemetry that lies is worse than
    telemetry that is absent (#2663)."""
    rule_usage.record_rule_outcome(
        user_id=7, rule_id=156, outcome=DEPARTED,
        source="mcp_rule_outcome", detail=reason or "",
    )
    assert captured == []


def test_an_unknown_outcome_is_never_written(captured):
    """Including the one somebody will reach for. There is no `ignored`
    event by design — see `record_rule_outcome` — and a caller inventing one
    must not get a row that reads as though the state were measurable."""
    for bogus in ("ignored", "skipped", "surfaced", "", "APPLIED "):
        rule_usage.record_rule_outcome(
            user_id=7, rule_id=156, outcome=bogus, source="mcp_rule_outcome"
        )
    assert captured == []


def test_an_outcome_row_is_one_row(captured):
    """A judgement is about one rule. Unlike a surfacing, which delivers a
    whole hint at once, there is no batch shape to get wrong here — asserted
    so that a later 'helpful' bulk variant has to change a test that says
    why."""
    rule_usage.record_rule_outcome(
        user_id=7, rule_id=1, outcome=APPLIED, source="mcp_rule_outcome"
    )
    [batch] = captured
    assert len(batch) == 1


# ── the four states, read off the aggregate ───────────────────────────────

def _usage(**kw):
    base = rule_usage.empty_rule_usage()
    base.update(kw)
    return base


def test_a_rule_surfaced_and_never_opened_is_unread():
    assert rule_usage.outcome_state(_usage(surfaced_count=4)) == rule_usage.UNREAD


def test_a_rule_opened_and_acted_on_is_followed():
    assert rule_usage.outcome_state(
        _usage(surfaced_count=4, pull_count=1, applied_count=1)
    ) == rule_usage.FOLLOWED


def test_a_rule_opened_and_departed_from_is_departed():
    assert rule_usage.outcome_state(
        _usage(surfaced_count=4, pull_count=1, departed_count=1)
    ) == rule_usage.DEPARTED_FROM


def test_a_rule_opened_and_never_acted_on_is_unacted() -> None:
    """THE STATE THAT DID NOT EXIST, and the reason for the milestone. Not
    "no data": the rule was surfaced, deliberately opened, and then left no
    trace of having mattered. Until now that was arithmetically identical to
    compliance, which is why nothing could report it."""
    assert rule_usage.outcome_state(
        _usage(surfaced_count=4, pull_count=2)
    ) == rule_usage.UNACTED


def test_unacted_and_followed_are_not_the_same_reading():
    """Stated as its own test because it IS the milestone in one line. If a
    change ever makes these two agree, the measurement is gone and every
    other test here would still pass."""
    opened_only = _usage(surfaced_count=4, pull_count=2)
    opened_and_applied = _usage(surfaced_count=4, pull_count=2, applied_count=1)
    assert rule_usage.outcome_state(opened_only) != rule_usage.outcome_state(
        opened_and_applied
    )


def test_a_departure_outranks_an_application():
    """A rule both applied and argued with is a rule someone argued with, and
    the argument is the half worth surfacing. Reporting it as plain
    compliance would bury the one row a reader most wants to see."""
    assert rule_usage.outcome_state(
        _usage(pull_count=3, applied_count=5, departed_count=1)
    ) == rule_usage.DEPARTED_FROM


def test_the_state_reads_the_aggregate_the_readout_already_returns():
    """`outcome_state` takes `usage_for_rules`' own shape, so the badge, the
    readout and any later session summary cannot disagree about what
    "followed" means — the drift #3246 found across the rules system."""
    assert rule_usage.outcome_state(rule_usage.empty_rule_usage()) == rule_usage.UNREAD