Detect syndication from text, not just from configured ownership

Connor's observation: two outlets running the same story can be caught by
comparing the text, without anyone having to know who owns whom.

Corroboration is now origin_count(), a union-find where two items share an
origin if EITHER they are in the same bucket, OR their text is near-identical
(normalised title match, or cosine >= 0.97). The rules are complementary:
buckets are same-owner/different-text (NYT filing five distinct AfD pieces),
text matching is different-owner/same-text (a local station reprinting wire).

On real data it fires exactly where it should: the vote-by-mail cluster
collapses 2 buckets -> 1 origin, catching the States Newsroom syndication
automatically -- the case previously fixed by hand in config -- and a
Smithsonian story collapses 3 -> 2, catching WCPO reprinting wire, which no
static bucket can express since WCPO is genuinely independent on local stories.
The eight Renoir-heist reports, each independently written, correctly stay at 8.
Across 91 multi-item clusters: 174 buckets -> 163 true origins.

Note this is the one place a lexical comparison is right. Clustering fails on
lexical methods because paraphrases share no wording, which is why the SimHash
stage was measured and deleted -- but syndicated copy is literally the same
bytes.

First formulation was wrong and the data caught it: summing syndication groups
and solo buckets produced 3 origins for a 2-bucket cluster. Syndication may only
ever reduce corroboration, which the union-find guarantees and a test asserts.

news/cli.py

@@ -51,7 +51,10 @@ def cmd_clusters(args) -> int:
     ).fetchall()
     for r in rows:
         buckets = cluster_mod.bucket_count(con, r["id"])
-        print(f"\n[{r['size']:>3} items, {buckets} independent bucket(s)] {r['lead']}")
+        origins = cluster_mod.origin_count(
+            con, r["id"], Config().clustering["syndication_threshold"])
+        synd = f", {buckets - origins} syndicated" if origins < buckets else ""
+        print(f"\n[{r['size']:>3} items, {origins} independent origin(s){synd}] {r['lead']}")
         if args.verbose:
             for m in con.execute(
                 """SELECT i.title, s.name FROM cluster_members cm

news/cluster.py

@@ -22,6 +22,7 @@ feeds importance, and it merges unrelated stories into one notification.
 
 from __future__ import annotations
 
+import re
 import sqlite3
 from dataclasses import dataclass
 from datetime import datetime, timedelta
@@ -33,6 +34,11 @@ from .util import stable_id
 
 THRESHOLD = 0.82
 WINDOW_HOURS = 48
+# Above this cosine, two items from different newsrooms are the same syndicated
+# copy rather than independent reporting. High on purpose: independently written
+# headlines about one event sit far below it (the eight Renoir-heist reports all
+# stayed separate), while wire reprints are effectively identical text.
+SYNDICATION_THRESHOLD = 0.97
 
 
 @dataclass(frozen=True)
@@ -142,12 +148,85 @@ def assign(
     return out
 
 
-def bucket_count(con: sqlite3.Connection, cluster_id: str) -> int:
+_PUNCT = re.compile(r"[^a-z0-9 ]")
-    """Distinct editorial-independence buckets in a cluster.
 
-    This, not raw member count, is the corroboration signal. Ten outlets
+
-    reprinting one wire story is one bucket; the old system counted it as ten
+def _normalise_title(title: str) -> str:
-    and then fired a priority-5 push on it.
+    return " ".join(_PUNCT.sub("", title.lower()).split())
+
+
+def origin_count(
+    con: sqlite3.Connection,
+    cluster_id: str,
+    syndication_threshold: float = SYNDICATION_THRESHOLD,
+) -> int:
+    """Number of independent newsrooms that originated reporting in a cluster.
+
+    This, not member count and not bucket count, is the corroboration signal.
+    Two items share an origin when EITHER holds:
+
+    1. Same `bucket` -- the same newsroom, however many feeds or distinct
+       stories it filed. Catches NYT running five different AfD pieces.
+    2. Near-identical text -- syndicated copy, whoever published it. Catches a
+       local station reprinting a wire dispatch, which no static bucket can
+       express: WCPO is genuinely independent on Cincinnati stories and a pure
+       reprinter on national ones.
+
+    Those two rules are complementary and both are needed: (1) is same-owner /
+    different-text, (2) is different-owner / same-text. Merges are transitive.
+
+    Note this is where a lexical comparison IS the right tool, unlike
+    clustering. Paraphrases of one story share almost no wording, which is why
+    the SimHash stage was measured and removed -- but syndicated copy is
+    literally the same bytes.
+    """
+    rows = con.execute(
+        """SELECT i.id, i.title, i.content_hash, s.bucket
+           FROM cluster_members cm
+           JOIN items i   ON i.id = cm.item_id
+           JOIN sources s ON s.id = i.source_id
+           WHERE cm.cluster_id = ?""",
+        (cluster_id,),
+    ).fetchall()
+    if not rows:
+        return 0
+
+    vectors: dict[str, np.ndarray] = {}
+    for row in rows:
+        got = con.execute(
+            "SELECT vec FROM embeddings WHERE content_hash=?", (row["content_hash"],)
+        ).fetchone()
+        if got is not None:
+            vectors[row["content_hash"]] = store.from_blob(got["vec"])
+
+    parent = list(range(len(rows)))
+
+    def find(a: int) -> int:
+        while parent[a] != a:
+            parent[a] = parent[parent[a]]
+            a = parent[a]
+        return a
+
+    titles = [_normalise_title(r["title"]) for r in rows]
+    for i in range(len(rows)):
+        for j in range(i + 1, len(rows)):
+            same = rows[i]["bucket"] == rows[j]["bucket"] or titles[i] == titles[j]
+            if not same:
+                vi = vectors.get(rows[i]["content_hash"])
+                vj = vectors.get(rows[j]["content_hash"])
+                if vi is not None and vj is not None:
+                    same = float(vi @ vj) >= syndication_threshold
+            if same:
+                parent[find(i)] = find(j)
+
+    return len({find(i) for i in range(len(rows))})
+
+
+def bucket_count(con: sqlite3.Connection, cluster_id: str) -> int:
+    """Distinct configured buckets, ignoring per-story syndication.
+
+    Kept for diagnostics: comparing it with `origin_count` shows how much
+    syndication a cluster contains. Routing should use `origin_count`.
     """
     row = con.execute(
         """SELECT COUNT(DISTINCT s.bucket) AS n

news/config.py

@@ -82,7 +82,11 @@ class Config:
     @property
     def clustering(self) -> dict[str, Any]:
         c = self.section("clustering")
-        return {"threshold": c.get("threshold", 0.82), "window_hours": c.get("window_hours", 48)}
+        return {
+            "threshold": c.get("threshold", 0.82),
+            "window_hours": c.get("window_hours", 48),
+            "syndication_threshold": c.get("syndication_threshold", 0.97),
+        }
 
     @property
     def scoring(self) -> dict[str, Any]:

news/pipeline.py

@@ -129,7 +129,8 @@ def run_cycle(con: sqlite3.Connection, cfg: Config, embedder: Embedder) -> Cycle
         clustering = cfg.clustering
         assignments = cluster.assign(
             con, list(pending), vectors,
-            threshold=clustering["threshold"], window_hours=clustering["window_hours"])
+            threshold=clustering["threshold"],
+            window_hours=clustering["window_hours"])
         report.clustered = len(assignments)
         report.new_clusters = sum(1 for a in assignments if a.created)
     return report

profile/config.toml

@@ -40,6 +40,12 @@ dim = 384
 threshold = 0.82
 window_hours = 48
 
+# Above this cosine, two items from DIFFERENT newsrooms are treated as the same
+# syndicated copy rather than independent corroboration. Deliberately high:
+# eight outlets each writing their own Renoir-heist headline stayed separate,
+# while a local station reprinting a wire dispatch collapsed correctly.
+syndication_threshold = 0.97
+
 [scoring]
 # Compare blocks by z-score WITHIN each block, never by raw cosine: cosine
 # magnitude is not comparable across blocks, and a plain max let one block win

tests/test_cluster.py

@@ -96,3 +96,60 @@ def test_syndicated_partners_in_one_bucket_count_once(con):
     assert out[0].cluster_id == out[1].cluster_id
     assert cluster.bucket_count(con, out[0].cluster_id) == 1, (
         "syndicated partners must not read as independent confirmation")
+
+
+def _cluster_of(con, items_spec, vec):
+    """Put every item in one cluster and return its id."""
+    rows = [add(con, iid, sid, when, chash) for iid, sid, when, chash in items_spec]
+    out = cluster.assign(con, rows, {c: vec for *_, c in items_spec})
+    return out[0].cluster_id
+
+
+def test_origin_count_collapses_syndicated_copy_across_newsrooms(con):
+    """Different owners, identical text: a wire reprint is not corroboration.
+
+    This is the case no static bucket can express -- a local station is
+    independent on local stories and a pure reprinter on national ones.
+    """
+    v = unit(1, 0, 0)
+    con.execute("INSERT OR REPLACE INTO embeddings (content_hash,dim,vec) VALUES (?,?,?)",
+                ("hw", 3, store.to_blob(v)))
+    rows = [add(con, "w1", "s1", "2026-01-01T00:00:00+00:00", "hw"),
+            add(con, "w2", "s2", "2026-01-01T00:20:00+00:00", "hw")]
+    # identical headline text across two different buckets
+    con.execute("UPDATE items SET title='Wire dispatch: minister resigns' WHERE id IN ('w1','w2')")
+    rows = con.execute("SELECT * FROM items WHERE id IN ('w1','w2') ORDER BY published_at").fetchall()
+    cid = cluster.assign(con, list(rows), {"hw": v})[0].cluster_id
+    assert cluster.bucket_count(con, cid) == 2
+    assert cluster.origin_count(con, cid) == 1
+
+
+def test_origin_count_keeps_independently_written_reports_separate(con):
+    """Same event, different newsrooms, different wording: real corroboration."""
+    # cosine ~0.90: above the 0.82 clustering threshold so they land in one
+    # cluster, below the 0.97 syndication threshold so they stay distinct
+    # origins. The first draft of this test used ~0.999 vectors and correctly
+    # merged them -- the test data was wrong, not the rule.
+    a, b = unit(1, 0, 0), unit(0.90, 0.436, 0)
+    for h, v in (("ha", a), ("hb", b)):
+        con.execute("INSERT OR REPLACE INTO embeddings (content_hash,dim,vec) VALUES (?,?,?)",
+                    (h, 3, store.to_blob(v)))
+    add(con, "r1", "s1", "2026-01-01T00:00:00+00:00", "ha")
+    add(con, "r2", "s2", "2026-01-01T00:30:00+00:00", "hb")
+    con.execute("UPDATE items SET title='Thieves steal two paintings' WHERE id='r1'")
+    con.execute("UPDATE items SET title='Renoir works taken in museum heist' WHERE id='r2'")
+    rows = con.execute("SELECT * FROM items WHERE id IN ('r1','r2') ORDER BY published_at").fetchall()
+    cid = cluster.assign(con, list(rows), {"ha": a, "hb": b})[0].cluster_id
+    assert cluster.origin_count(con, cid) == 2
+
+
+def test_origin_count_never_exceeds_bucket_count(con):
+    """Syndication detection may only ever reduce corroboration."""
+    v = unit(1, 0, 0)
+    con.execute("INSERT OR REPLACE INTO embeddings (content_hash,dim,vec) VALUES (?,?,?)",
+                ("hx", 3, store.to_blob(v)))
+    specs = [("x1", "s1", "2026-01-01T00:00:00+00:00", "hx"),
+             ("x2", "s2", "2026-01-01T00:10:00+00:00", "hx"),
+             ("x3", "s1", "2026-01-01T00:20:00+00:00", "hx")]
+    cid = _cluster_of(con, specs, v)
+    assert cluster.origin_count(con, cid) <= cluster.bucket_count(con, cid)

tests/test_config.py

@@ -64,7 +64,8 @@ def test_machinery_values_are_read_from_config(cfgdir):
     assert cfg.http["timeout_secs"] == 7
     assert cfg.http["concurrency"] == 3
     assert cfg.embedding == {"model": "intfloat/multilingual-e5-small", "dim": 512}
-    assert cfg.clustering == {"threshold": 0.5, "window_hours": 6}
+    assert cfg.clustering["threshold"] == 0.5
+    assert cfg.clustering["window_hours"] == 6
     assert cfg.scoring["veto_absolute"] == 0.99