news/tests/test_place_gate.py

"""Place gating.

Embeddings model meaning well and named entities poorly: "Italian politics" and
"Australian politics" sit almost on top of each other. Measured on live data,
66% of the winners of a place-based block named no place at all -- Glastonbury
and Anthony Albanese were scoring as `italy-mexico`.

Place is identity, not meaning, so it is matched lexically.
"""

from __future__ import annotations

import numpy as np
import pytest

from news import store
from news.score import ProfileVectors, _mentions, score_clusters


def test_word_boundaries_not_substrings():
    """The lesson from "nfl" matching co-NFL-ict, which vetoed every Iran story."""
    assert _mentions("meloni visits rome today", ["rome"])
    assert not _mentions("open the chrome browser", ["rome"])
    assert not _mentions("efforts to normalise ties", ["mali"])
    assert _mentions("fighting spreads in mali", ["mali"])


class _FakeVectors:
    """Two blocks in different groups: one gated on a place, one ungated."""

    def __init__(self):
        near = np.array([[1.0, 0.0]], dtype=np.float32)
        self.blocks = {
            "interest": [
                ("italy-mexico", near, {"id": "italy-mexico", "weight": 1.0,
                                        "requires_any": ["italy", "mexico"]}),
                ("urbanism", np.array([[0.0, 1.0]], dtype=np.float32),
                 {"id": "urbanism", "weight": 1.0}),
            ],
            "veto": [], "demote": [], "importance": [
                ("x", np.array([[0.0, 0.0]], dtype=np.float32), {"id": "x", "weight": 1.0})],
        }

    def raw_matrix(self, kind, vecs):
        return {bid: (vecs @ m.T).max(axis=1) for bid, m, _ in self.blocks[kind]}

    def weight(self, kind, bid):
        return 1.0

    def group(self, kind, bid):
        return bid

    def block(self, kind, bid):
        return next(b for i, _, b in self.blocks[kind] if i == bid)


@pytest.fixture
def con(tmp_path):
    c = store.connect(tmp_path / "g.db")
    store.upsert_source(c, id="s1", name="A", url="http://a", kind="rss", bucket="a",
                        weight=1.0, prominence=0, enabled=1, poll_interval_secs=900)
    return c


def _add(con, cid, title):
    when = "2026-01-01T00:00:00+00:00"
    con.execute("""INSERT INTO items (id,source_id,title,summary,url,canonical_url,
                   published_at,first_seen_at,content_hash) VALUES (?,?,?,'',?,?,?,?,?)""",
                (cid, "s1", title, f"http://x/{cid}", f"http://x/{cid}", when, when, cid))
    con.execute("""INSERT INTO clusters (id,centroid,size,first_seen_at,last_seen_at,lead_item_id)
                   VALUES (?,?,1,?,?,?)""",
                (cid, store.to_blob(np.array([1.0, 0.0], dtype=np.float32)), when, when, cid))
    con.execute("INSERT INTO cluster_members VALUES (?,?,?)", (cid, cid, when))


CFG = {"veto_absolute": 0.9, "veto_margin_over_interest": 9.0,
       "corroboration_pickup_weight": 0.3, "corroboration_depth_weight": 0.2}
CLUST = {"syndication_threshold": 0.97}


def test_a_gated_block_cannot_win_without_a_mention(con):
    """Both items point at the italy-mexico probe; only one names a place."""
    _add(con, "yes", "Meloni's government becomes Italy's longest-lasting")
    _add(con, "no", "Anthony Albanese held an undeclared honorary membership")
    scores = {s.cluster_id: s for s in score_clusters(con, _FakeVectors(), CFG, CLUST)}
    assert scores["yes"].interest.block == "italy-mexico"
    assert scores["no"].interest.block != "italy-mexico"


def test_gate_survives_normalisation(con):
    """Regression for a real bug.

    The gate was first applied by writing a sentinel into the RAW score before
    normalisation. When most items are gated that sentinel becomes the block's
    own mean, so a gated item z-scored to roughly zero and could still win --
    the normalisation erased the gate. With 30 gated items against 1 mention,
    a pre-normalisation gate lets the gated ones through.
    """
    _add(con, "mention", "Sheinbaum speaks in Mexico")
    for i in range(30):
        _add(con, f"other{i}", f"Unrelated story number {i}")
    scores = score_clusters(con, _FakeVectors(), CFG, CLUST)
    winners = [s for s in scores if s.interest.block == "italy-mexico"]
    assert [s.cluster_id for s in winners] == ["mention"], (
        "only the item naming a place may win the gated block")


def test_one_block_in_a_group_is_enough_to_clear_the_gate(tmp_path):
    """Regression: the gate was a union across a group, so a group of blocks
    with disjoint gazetteers gated everything.

    `local` holds cincinnati, kentucky-louisville and ohio-state-politics. A
    Cincinnati story does not say "Kentucky", so kentucky-louisville gated it
    and the whole group -- including the block that matched at 0.697, the
    highest score on either axis -- was set to -inf. The block took 0 of 196
    digest slots in a week because of this line.
    """
    import numpy as np
    from news import store
    from news.score import score_clusters

    con = store.connect(tmp_path / "g.db")
    store.upsert_source(con, id="s1", name="WCPO Cincinnati", url="http://w",
                        kind="rss", bucket="wcpo", weight=1.0, prominence=0,
                        enabled=1, poll_interval_secs=900)

    profile = {"interest": [
        {"id": "cincinnati", "group": "local", "weight": 1.0,
         "requires_any": ["cincinnati"],
         "probes": ["Cincinnati city council votes on a local ordinance."]},
        {"id": "kentucky", "group": "local", "weight": 1.0,
         "requires_any": ["kentucky"],
         "probes": ["Kentucky state politics and legislation in Frankfort."]},
    ], "veto": [], "demote": [], "importance": [
        {"id": "x", "weight": 1.0, "probes": ["A major disaster kills many people."]}]}

    titles = ["Cincinnati city council votes on a stormwater ordinance",
              "Kentucky legislature debates a new budget in Frankfort",
              "Sydney transport authority approves a new ferry timetable"]
    for n, title in enumerate(titles):
        when = f"2026-09-1{n + 1}T00:00:00+00:00"
        con.execute("""INSERT INTO items (id,source_id,title,summary,url,canonical_url,
                       published_at,first_seen_at,content_hash) VALUES (?,?,?,'',?,?,?,?,?)""",
                    (f"i{n}", "s1", title, f"http://x/{n}", f"http://x/{n}", when, when, f"h{n}"))

    from news.embed import Embedder
    from news.score import ProfileVectors
    emb = Embedder.from_config(__import__("news.config", fromlist=["Config"]).Config())
    vecs = emb.encode(titles)
    for n in range(len(titles)):
        con.execute("""INSERT INTO clusters (id,centroid,size,first_seen_at,last_seen_at,
                       lead_item_id) VALUES (?,?,1,?,?,?)""",
                    (f"c{n}", store.to_blob(vecs[n]),
                     f"2026-09-1{n + 1}T00:00:00+00:00", f"2026-09-1{n + 1}T00:00:00+00:00",
                     f"i{n}"))
        con.execute("INSERT INTO cluster_members VALUES (?,?,?)",
                    (f"c{n}", f"i{n}", f"2026-09-1{n + 1}T00:00:00+00:00"))

    pv = ProfileVectors(con, profile, emb)
    scored = {s.cluster_id: s for s in score_clusters(
        con, pv, {"normalize": "zscore-per-block", "veto_absolute": 0.6,
                  "veto_margin_over_interest": 0.3, "corroboration_pickup_weight": 0.3,
                  "corroboration_depth_weight": 0.2, "importance_weights": {}},
        {"syndication_threshold": 0.97})}

    import math
    assert not math.isinf(scored["c0"].interest_score), (
        "the Cincinnati story was gated even though it names Cincinnati")
    assert not math.isinf(scored["c1"].interest_score), (
        "the Kentucky story was gated even though it names Kentucky")
    assert math.isinf(scored["c2"].interest_score) and scored["c2"].interest_score < 0, (
        "the Sydney story names neither place and must still be gated")