news/tests/test_route.py

"""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


def _seen_on(con, day, *cluster_ids):
    """Record a prior digest appearance."""
    for rank, cid in enumerate(cluster_ids):
        con.execute("""INSERT INTO digest_items (day,cluster_id,rank,score,reason)
                       VALUES (?,?,?,1.0,'digest')""", (day.isoformat(), cid, rank))


def test_only_the_strongest_repeats_come_back(con):
    """A story stays eligible while fresh reports keep joining its cluster, so
    without a cap the digest becomes mostly re-runs -- 9 of 28 on a real day."""
    yesterday = (NOW - timedelta(days=1)).date()
    _seen_on(con, yesterday, *[f"old{i}" for i in range(6)])

    # Six returning stories at descending strength, plus fresh material. All
    # below interest_threshold, or they would be notified rather than digested.
    scores = [mk(f"old{i}", interest=1.9 - i * 0.1) for i in range(6)]
    scores += [mk(f"new{i}", interest=1.0) for i in range(10)]
    out = route(con, scores, {**ROUTING, "digest_max_repeats": 3}, DELIVERY, NOW)

    digested = {d.cluster_id for d in out if d.lane == DIGEST}
    assert {"old0", "old1", "old2"} <= digested, "the strongest repeats must return"
    for cid in ("old3", "old4", "old5"):
        assert cid not in digested, f"{cid} is a fourth-or-later repeat"
    # Freed slots go to stories not yet seen, not to nothing.
    assert len([c for c in digested if c.startswith("new")]) == 10


def test_a_capped_repeat_says_why(con):
    _seen_on(con, (NOW - timedelta(days=1)).date(), "a", "b")
    out = route(con, [mk("a", interest=1.9), mk("b", interest=1.8)],
                {**ROUTING, "digest_max_repeats": 1}, DELIVERY, NOW)
    reasons = {d.cluster_id: d.reason for d in out}
    assert reasons["b"] == "repeat-capped"


def test_repeats_are_ranked_by_best_match_not_interest_alone(con):
    """A repeat earns its slot on either axis: a story you are only mildly
    interested in but which has escalated must still get through."""
    _seen_on(con, (NOW - timedelta(days=1)).date(), "quiet", "escalating")
    scores = [mk("quiet", interest=1.5, importance=0.0),
              mk("escalating", interest=0.6, importance=1.95)]
    out = route(con, scores, {**ROUTING, "digest_max_repeats": 1}, DELIVERY, NOW)
    kept = {d.cluster_id for d in out if d.lane == DIGEST}
    assert kept == {"escalating"}, (
        "ranked by interest alone, the quiet story would have taken the slot")


def test_the_importance_allowance_cannot_smuggle_repeats_past_the_cap(con):
    """The group cap failed this exact way once: a bounded allowance applied
    after the fact swallowed the rule it was meant to respect."""
    _seen_on(con, (NOW - timedelta(days=1)).date(), *[f"r{i}" for i in range(6)])
    scores = [mk(f"r{i}", interest=0.6, importance=9.0 - i * 0.1) for i in range(6)]
    out = route(con, scores, {**ROUTING, "digest_max_repeats": 2,
                              "digest_importance_slots": 6}, DELIVERY, NOW)
    assert sum(1 for d in out if d.lane == DIGEST) == 2


def test_a_repeat_cannot_return_through_a_serendipity_slot(con):
    _seen_on(con, (NOW - timedelta(days=1)).date(), "off1", "off2", "off3")
    delivery = {**DELIVERY, "serendipity_slots": 2}
    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),
               mk("off3", interest=-3.0, importance=0.8)]
    out = route(con, scores, {**ROUTING, "digest_max_items": 5, "digest_max_repeats": 1},
                delivery, NOW)
    picked = {d.cluster_id for d in out if d.reason == "serendipity"}
    assert picked <= {"off1"}, f"a capped repeat came back as serendipity: {picked}"


def test_rerunning_the_digest_does_not_treat_its_own_rows_as_repeats(con):
    """`day < today`: without it, a second run on the same day would see this
    morning's rows and cap the whole digest to three stories."""
    _seen_on(con, NOW.date(), *[f"t{i}" for i in range(6)])
    scores = [mk(f"t{i}", interest=1.0) for i in range(6)]
    out = route(con, scores, {**ROUTING, "digest_max_repeats": 3}, DELIVERY, NOW)
    assert sum(1 for d in out if d.lane == DIGEST) == 6


def test_a_story_returning_after_the_lookback_counts_as_new(con):
    _seen_on(con, (NOW - timedelta(days=10)).date(), "a", "b", "c", "d")
    scores = [mk(c, interest=1.0) for c in ("a", "b", "c", "d")]
    out = route(con, scores, {**ROUTING, "digest_max_repeats": 1,
                              "digest_repeat_lookback_days": 3}, DELIVERY, NOW)
    assert sum(1 for d in out if d.lane == DIGEST) == 4