Phase 2: ingest, embed, cluster

Pipeline: fetch -> normalise -> embed (CPU) -> cluster. `news poll` runs a
cycle; `news sources` and `news clusters` inspect the result. Replay and the
live service share run_cycle so tuning and production cannot diverge -- the old
system's replay harness hardcoded default params and could never sweep the
thresholds it existed to tune.

Calibrated, not guessed:

- Cluster threshold 0.82, from reading real output. 0.70 merged "Solheim Cup
  capsules" with "FIBA Women's World Cup Glance"; 0.82 gives clean groups (76
  items on the AfD Saxony-Anhalt election, 27 on the Nepal tunnel rescue). The
  old system produced 515 clusters of which 514 were singletons.
- Corroboration counts DISTINCT buckets, not feeds, so ten outlets reprinting
  one wire story is one confirmation rather than ten.

Removed after measurement: a SimHash near-duplicate stage. It gave +1.2 bits of
separation on titles and -0.2 with summaries -- news duplicates are paraphrases,
which is the same lexical trap that sank the old Jaccard clustering.

Sources: all 39 fetched against the live web, not assumed. Dropped CBC (every
feed URL times out from this host) and Tech Policy Press (serves HTML at every
documented feed path). Fixed WVXU (index.rss is an empty stub; local-news.rss
works). Reddit moved to .rss because .json now 403s without OAuth, and its
intervals raised because it 429s after three quick requests. Deutsche Welle is
back: it ships RSS 1.0/RDF, which feedparser reads and the old hand-rolled
quick-xml parser could not.

tests/test_cluster.py

@@ -0,0 +1,81 @@
+"""Clustering behaviour, using synthetic vectors so no model download is needed."""
+
+import numpy as np
+import pytest
+
+from news import cluster, store
+from news.util import stable_id
+
+
+@pytest.fixture
+def con(tmp_path):
+    c = store.connect(tmp_path / "t.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)
+    store.upsert_source(
+        c, id="s2", name="B", url="http://b", kind="rss", bucket="b",
+        weight=1.0, prominence=0, enabled=1, poll_interval_secs=900)
+    return c
+
+
+def add(con, item_id, source_id, when, chash):
+    con.execute(
+        """INSERT INTO items (id,source_id,title,summary,url,canonical_url,
+                              published_at,first_seen_at,content_hash)
+           VALUES (?,?,?,'',?,?,?,?,?)""",
+        (item_id, source_id, f"t-{item_id}", f"http://x/{item_id}",
+         f"http://x/{item_id}", when, when, chash))
+    return con.execute("SELECT * FROM items WHERE id=?", (item_id,)).fetchone()
+
+
+def unit(*v):
+    a = np.array(v, dtype=np.float32)
+    return a / np.linalg.norm(a)
+
+
+def test_similar_items_merge_and_distinct_ones_do_not(con):
+    near_a, near_b = unit(1, 0.02, 0), unit(1, 0.06, 0)   # cosine ~0.999
+    far = unit(0, 1, 0)                                    # orthogonal
+    rows = [add(con, "i1", "s1", "2026-01-01T00:00:00+00:00", "h1"),
+            add(con, "i2", "s2", "2026-01-01T01:00:00+00:00", "h2"),
+            add(con, "i3", "s1", "2026-01-01T02:00:00+00:00", "h3")]
+    out = cluster.assign(con, rows, {"h1": near_a, "h2": near_b, "h3": far})
+    by_item = {a.item_id: a.cluster_id for a in out}
+    assert by_item["i1"] == by_item["i2"], "paraphrase-near items must merge"
+    assert by_item["i3"] != by_item["i1"], "unrelated item must not merge"
+
+
+def test_items_outside_the_window_do_not_merge(con):
+    v = unit(1, 0, 0)
+    rows = [add(con, "i1", "s1", "2026-01-01T00:00:00+00:00", "h1"),
+            add(con, "i2", "s2", "2026-01-05T00:00:00+00:00", "h2")]  # 4 days later
+    out = cluster.assign(con, rows, {"h1": v, "h2": v})
+    assert out[0].cluster_id != out[1].cluster_id
+
+
+def test_bucket_count_counts_newsrooms_not_feeds(con):
+    """Two feeds from one newsroom are one independent confirmation, not two."""
+    v = unit(1, 0, 0)
+    store.upsert_source(
+        con, id="s3", name="A2", url="http://a2", kind="rss", bucket="a",
+        weight=1.0, prominence=0, enabled=1, poll_interval_secs=900)
+    rows = [add(con, "i1", "s1", "2026-01-01T00:00:00+00:00", "h1"),
+            add(con, "i2", "s3", "2026-01-01T01:00:00+00:00", "h2")]
+    out = cluster.assign(con, rows, {"h1": v, "h2": v})
+    assert out[0].cluster_id == out[1].cluster_id
+    assert cluster.bucket_count(con, out[0].cluster_id) == 1
+
+    rows2 = [add(con, "i3", "s2", "2026-01-01T02:00:00+00:00", "h3")]
+    cluster.assign(con, rows2, {"h3": v})
+    assert cluster.bucket_count(con, out[0].cluster_id) == 2
+
+
+def test_assignment_is_idempotent(con):
+    v = unit(1, 0, 0)
+    rows = [add(con, "i1", "s1", "2026-01-01T00:00:00+00:00", "h1")]
+    a = cluster.assign(con, rows, {"h1": v})
+    b = cluster.assign(con, rows, {"h1": v})
+    assert a[0].cluster_id == b[0].cluster_id
+    n = con.execute("SELECT COUNT(*) c FROM cluster_members").fetchone()["c"]
+    assert n == 1, "re-assigning an item must not duplicate membership"