"""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 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, "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()