"""Routing decisions. The most important assertion here is that a veto blocks BOTH lanes. In the old system the bypass lane never consulted relevance (gate.rs:302), so a blocklisted item could still fire at priority 5 -- which is how a baseball recap could interrupt someone who had blocklisted sports. """ from __future__ import annotations from datetime import UTC, datetime, timedelta import pytest from news import store from news.cluster import Corroboration from news.route import DIGEST, DROP, IMPORTANT, INTEREST, in_quiet_hours, route from news.score import BlockHit, ClusterScore NOW = datetime(2026, 9, 8, 14, 0, tzinfo=UTC) # daytime NIGHT = datetime(2026, 9, 8, 23, 30, tzinfo=UTC) # inside 22:00-07:00 ROUTING = { "interest_per_day": 2, "important_per_day": 1, "important_per_night": 1, "interest_threshold": 2.0, "important_threshold": 2.0, "digest_floor": 0.5, "hn_min_score": 150, "max_age_hours": 48, "digest_max_items": 25, "serendipity_max_interest": 0.5, } DELIVERY = { "quiet_hours": {"start": "22:00", "end": "07:00"}, "importance_may_wake": True, "conflict_general_requires_importance": True, "serendipity_slots": 0, } def mk(cid, *, interest=0.0, importance=0.0, veto=None, block="urbanism", imp_block="war-escalation", ext=None, origins=2, last_seen=NOW): return ClusterScore( cluster_id=cid, lead_title=cid, last_seen=last_seen, interest=BlockHit(block, interest, 0.5), importance=BlockHit(imp_block, importance, 0.5), corroboration=Corroboration(origins, 0, 0, float(origins)), velocity=0.0, prominence=0, ext_score=ext, veto=veto, demote=None, demote_penalty=0.0, interest_score=interest, importance_score=importance) @pytest.fixture def con(tmp_path): return store.connect(tmp_path / "r.db") def lanes(decisions): return {d.cluster_id: d.lane for d in decisions} def test_veto_blocks_the_importance_lane_too(con): """The v0.2 bug: bypass ignored relevance, so vetoed items still fired.""" out = route(con, [mk("v", interest=5.0, importance=9.0, veto="us-big-three-sports")], ROUTING, DELIVERY, NOW) assert lanes(out)["v"] == DROP assert out[0].reason.startswith("veto:") def test_high_interest_alerts_during_the_day(con): out = route(con, [mk("a", interest=3.0)], ROUTING, DELIVERY, NOW) assert lanes(out)["a"] == INTEREST def test_interest_never_overrides_quiet_hours(con): out = route(con, [mk("a", interest=9.0)], ROUTING, DELIVERY, NIGHT) assert lanes(out)["a"] == DIGEST assert out[0].reason == "quiet-hours" def test_importance_may_wake_but_only_within_the_nightly_cap(con): scores = [mk("i1", importance=9.0), mk("i2", importance=8.0)] out = lanes(route(con, scores, ROUTING, DELIVERY, NIGHT)) assert out["i1"] == IMPORTANT, "the top importance item may wake" assert out["i2"] == DIGEST, "the nightly cap is 1" def test_importance_cannot_wake_when_disabled(con): delivery = {**DELIVERY, "importance_may_wake": False} out = route(con, [mk("i", importance=9.0)], ROUTING, delivery, NIGHT) assert lanes(out)["i"] == DIGEST def test_daily_budget_is_spent_on_the_strongest_candidates(con): """Budget must go to the best items, not to whichever arrived first.""" scores = [mk("weak", interest=2.1), mk("strong", interest=8.0), mk("mid", interest=4.0)] out = lanes(route(con, scores, {**ROUTING, "interest_per_day": 1}, DELIVERY, NOW)) assert out["strong"] == INTEREST assert out["weak"] == DIGEST and out["mid"] == DIGEST def test_stale_clusters_never_notify(con): old = NOW - timedelta(days=5) out = route(con, [mk("s", interest=9.0, last_seen=old)], ROUTING, DELIVERY, NOW) assert lanes(out)["s"] == DROP assert out[0].reason == "stale" def test_immature_aggregator_items_are_held(con): """An HN post at 3 points is noise; it is re-scored as its score climbs.""" out = route(con, [mk("hn", interest=9.0, ext=5, origins=1)], ROUTING, DELIVERY, NOW) assert lanes(out)["hn"] == DROP assert out[0].reason == "aggregator-immature" ripe = route(con, [mk("hn", interest=9.0, ext=400, origins=1)], ROUTING, DELIVERY, NOW) assert lanes(ripe)["hn"] == INTEREST def test_general_conflict_needs_importance_to_interrupt(con): """Connor: war interest is low, but all wars are important.""" routine = mk("c", interest=9.0, importance=0.1, block="conflict-general") assert lanes(route(con, [routine], ROUTING, DELIVERY, NOW))["c"] == DIGEST major = mk("c", interest=9.0, importance=5.0, block="conflict-general") assert lanes(route(con, [major], ROUTING, DELIVERY, NOW))["c"] == IMPORTANT def test_digest_is_capped_by_count(con): scores = [mk(f"d{i}", interest=1.0) for i in range(40)] out = lanes(route(con, scores, {**ROUTING, "digest_max_items": 5}, DELIVERY, NOW)) assert sum(1 for v in out.values() if v == DIGEST) == 5 def test_serendipity_fills_from_outside_the_profile(con): delivery = {**DELIVERY, "serendipity_slots": 2} # On-profile items must outrank the off-profile ones and fill the cap, # otherwise the off-profile items make the digest on merit and the reserved # slots have nothing to do. (An earlier version of this test got that # backwards and failed against correct code.) scores = [mk(f"on{i}", interest=1.9, importance=1.9) for i in range(30)] scores += [mk("off1", interest=-1.0, importance=1.0), mk("off2", interest=-2.0, importance=0.9)] out = route(con, scores, {**ROUTING, "digest_max_items": 5}, delivery, NOW) picked = {d.cluster_id for d in out if d.reason == "serendipity"} assert picked == {"off1", "off2"} def test_already_notified_clusters_are_not_repeated(con): con.execute( "INSERT INTO notification_log (cluster_id,lane,score,created_at) VALUES (?,?,?,?)", ("a", INTEREST, 5.0, NOW.isoformat())) out = route(con, [mk("a", interest=9.0)], ROUTING, DELIVERY, NOW) assert lanes(out)["a"] == DROP assert out[0].reason == "already-notified" @pytest.mark.parametrize("hhmm,expected", [ ("23:30", True), ("02:00", True), ("06:59", True), ("07:00", False), ("14:00", False), ("21:59", False), ("22:00", True)]) def test_quiet_hours_wrap_midnight(hhmm, expected): h, m = map(int, hhmm.split(":")) now = datetime(2026, 9, 8, h, m, tzinfo=UTC) assert in_quiet_hours(now, {"start": "22:00", "end": "07:00"}) is expected # --- timezone and pacing ----------------------------------------------------- DELIVERY_TZ = {**DELIVERY, "timezone": "America/Kentucky/Louisville"} def test_quiet_hours_use_local_time_not_utc(con): """Regression for the 3:39am avalanche. 22:00-07:00 was compared against UTC hours, so notifications were silenced from 18:00 local and the day's whole budget was released at 03:00 local. """ # 23:00 UTC == 19:00 EDT: awake locally, but "quiet" if you compare UTC. evening = datetime(2026, 9, 9, 23, 0, tzinfo=UTC) out = route(con, [mk("a", interest=9.0)], ROUTING, DELIVERY_TZ, evening) assert lanes(out)["a"] == INTEREST, "19:00 local is not quiet hours" # 03:30 UTC == 23:30 EDT: genuinely quiet locally. night = datetime(2026, 9, 10, 3, 30, tzinfo=UTC) out = route(con, [mk("b", interest=9.0)], ROUTING, DELIVERY_TZ, night) assert lanes(out)["b"] == DIGEST assert out[0].reason == "quiet-hours" def test_an_unknown_timezone_does_not_stop_delivery(con): delivery = {**DELIVERY, "timezone": "Mars/Olympus_Mons"} out = route(con, [mk("a", interest=9.0)], ROUTING, delivery, NOW) assert lanes(out)["a"] in (INTEREST, DIGEST) # decided, not crashed def test_only_one_notification_per_lane_per_cycle(con): """Budgets are not a rate. Six notifications once arrived in the same minute.""" scores = [mk(f"i{n}", interest=9.0 - n) for n in range(4)] out = lanes(route(con, scores, {**ROUTING, "interest_per_day": 4}, DELIVERY_TZ, NOW)) assert sum(1 for v in out.values() if v == INTEREST) == 1 assert out["i0"] == INTEREST, "the strongest candidate takes the slot" def test_a_second_send_waits_for_the_gap(con): con.execute( "INSERT INTO notification_log (cluster_id,lane,score,created_at) VALUES (?,?,?,?)", ("earlier", INTEREST, 5.0, (NOW - timedelta(minutes=10)).isoformat())) out = route(con, [mk("a", interest=9.0)], ROUTING, DELIVERY_TZ, NOW) assert lanes(out)["a"] == DIGEST assert out[0].reason == "interest-too-soon" # Once the gap has elapsed it goes through. later = NOW + timedelta(minutes=80) out2 = route(con, [mk("a", interest=9.0)], ROUTING, DELIVERY_TZ, later) assert lanes(out2)["a"] == INTEREST