"""Rule usage telemetry — the parts that need no database. 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 ───────────────────────────────────────────────────────────── # Integration: a real GROUP BY over a real table. The unit tests above cover # the WRITE path and the zero shape; the aggregate the rule list renders needs # a real database. @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. 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 ──────────────────────────────────────────────────── # # What these guard is a distinction, not a payload. Without outcomes, a rule # read and obeyed and a rule read and ignored leave byte-identical telemetry, # so the readout cannot name the failure. 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.""" 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 UNACTED STATE. Not "no data": the rule was surfaced, deliberately opened, and then left no trace of having mattered. Without this state it is arithmetically identical to compliance, and nothing can 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.""" assert rule_usage.outcome_state(rule_usage.empty_rule_usage()) == rule_usage.UNREAD