Tiered corroboration: origins, pickups and depth scored separately

Connor's point: "several outlets carried this" covers situations that carry
different amounts of information, and origin_count() flattened them all to one
number. Splitting on same/different newsroom and same/different text gives four
cases:

  same newsroom, same text  -> pure duplicate (BBC World + BBC top). Zero
                               signal; the front-page placement it does carry is
                               already captured by sources.prominence.
  same newsroom, diff text  -> DEPTH. A newsroom filing four pieces on one story
                               is a real editorial judgement, previously
                               discarded entirely.
  diff newsroom, same text  -> PICKUPS. Syndicated copy; each carrier chose to
                               run it. Weak but non-zero.
  diff newsroom, diff text  -> ORIGINS. Independent judgement and independent
                               newsgathering. Strongest.

score = origins + 0.3*log1p(pickups) + 0.2*log1p(depth), weights in config.
Origins stay dominant; the others break ties among the many two- and
three-origin clusters, and log damping stops a high-volume feed inflating depth.

Measuring first changed the design. Depth outnumbers pickups 78 to 11 across 91
real clusters, because these sources are mostly national outlets writing their
own copy rather than local papers running wire -- so the more valuable half was
the one not originally proposed. UK settlements scores 10.55 from 10 origins
plus 15 same-newsroom pieces; AfD scores 5.55 from 5 origins plus 15.

Tests assert the four-case ordering directly, and that twenty pieces from one
newsroom still rank below two independent ones. Both needed a helper generating
vectors with exact pairwise cosine: the band between the 0.82 clustering
threshold and the 0.97 syndication threshold is narrow, and two earlier drafts
of these tests used ad-hoc vectors that fell outside it and asserted nothing.

news/cli.py

@@ -50,11 +50,15 @@ def cmd_clusters(args) -> int:
         (args.min_size, args.limit),
     ).fetchall()
     for r in rows:
-        buckets = cluster_mod.bucket_count(con, r["id"])
-        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']}")
+        sc = cfg.scoring
+        co = cluster_mod.corroboration(
+            con, r["id"], cfg.clustering["syndication_threshold"],
+            sc["corroboration_pickup_weight"], sc["corroboration_depth_weight"])
+        extra = "".join(
+            (f", {co.pickups} syndicated pickup(s)" if co.pickups else ""),
+        ) + (f", {co.depth} extra piece(s) same newsroom" if co.depth else "")
+        print(f"\n[{r['size']:>3} items | corroboration {co.score:.2f} "
+              f"= {co.origins} origin(s){extra}] {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 math
 import re
 import sqlite3
 from dataclasses import dataclass
@@ -39,6 +40,9 @@ WINDOW_HOURS = 48
 # 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
+# Tiebreaker weights. Small on purpose: origins must stay the dominant term.
+PICKUP_WEIGHT = 0.3
+DEPTH_WEIGHT = 0.2
 
 
 @dataclass(frozen=True)
@@ -155,33 +159,52 @@ def _normalise_title(title: str) -> str:
     return " ".join(_PUNCT.sub("", title.lower()).split())
 
 
-def origin_count(
+@dataclass(frozen=True)
+class Corroboration:
+    """How much independent judgement a cluster represents.
+
+    Not one number, because "several outlets carried this" covers four
+    situations that mean different things. Splitting on same/different newsroom
+    and same/different text:
+
+        same newsroom,  same text  -> pure duplicate (BBC World + BBC top).
+                                      Zero information. The editorial signal
+                                      there is the front-page placement, which
+                                      `sources.prominence` captures separately.
+        same newsroom,  diff text  -> DEPTH. A newsroom filing four pieces on
+                                      one story is a real judgement of
+                                      importance.
+        diff newsroom,  same text  -> PICKUPS. Syndicated copy. Each carrier
+                                      chose to run it: weak signal, not zero.
+        diff newsroom,  diff text  -> ORIGINS. Independent judgement AND
+                                      independent newsgathering. Strongest.
+
+    Measured on a real 91-cluster corpus: 163 origins, 90 depth, 11 pickups.
+    Depth dominates because these sources are mostly national outlets writing
+    their own copy; pickups would matter far more with local papers on the wire.
+    """
+
+    origins: int
+    pickups: int
+    depth: int
+    score: float
+
+
+def corroboration(
     con: sqlite3.Connection,
     cluster_id: str,
     syndication_threshold: float = SYNDICATION_THRESHOLD,
-) -> int:
-    """Number of independent newsrooms that originated reporting in a cluster.
+    pickup_weight: float = PICKUP_WEIGHT,
+    depth_weight: float = DEPTH_WEIGHT,
+) -> Corroboration:
+    """Score = origins + w_p*log1p(pickups) + w_d*log1p(depth).
 
-    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.
+    Origins dominate by design; the other two break ties among the many
+    two- and three-origin clusters. The log damping stops a high-volume feed
+    inflating `depth` by publishing many near-identical pieces.
     """
     rows = con.execute(
-        """SELECT i.id, i.title, i.content_hash, s.bucket
+        """SELECT 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
@@ -189,7 +212,7 @@ def origin_count(
         (cluster_id,),
     ).fetchall()
     if not rows:
-        return 0
+        return Corroboration(0, 0, 0, 0.0)
 
     vectors: dict[str, np.ndarray] = {}
     for row in rows:
@@ -199,7 +222,17 @@ def origin_count(
         if got is not None:
             vectors[row["content_hash"]] = store.from_blob(got["vec"])
 
-    parent = list(range(len(rows)))
+    n = len(rows)
+    titles = [_normalise_title(r["title"]) for r in rows]
+
+    def same_text(i: int, j: int) -> bool:
+        if titles[i] == titles[j]:
+            return True
+        vi, vj = vectors.get(rows[i]["content_hash"]), vectors.get(rows[j]["content_hash"])
+        return vi is not None and vj is not None and float(vi @ vj) >= syndication_threshold
+
+    def components(pred) -> list[int]:
+        parent = list(range(n))
 
         def find(a: int) -> int:
             while parent[a] != a:
@@ -207,19 +240,43 @@ def origin_count(
                 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:
+        for i in range(n):
+            for j in range(i + 1, n):
+                if pred(i, j):
                     parent[find(i)] = find(j)
+        return [find(i) for i in range(n)]
 
-    return len({find(i) for i in range(len(rows))})
+    # Text groups: items that are the same piece of writing, wherever published.
+    text_group = components(same_text)
+    # Origins: same newsroom OR same writing.
+    origin = components(lambda i, j: rows[i]["bucket"] == rows[j]["bucket"] or same_text(i, j))
+
+    origins = len(set(origin))
+
+    # Pickups: for each piece of writing carried by more than one newsroom,
+    # every carrier beyond the first.
+    buckets_per_text: dict[int, set[str]] = {}
+    for idx, g in enumerate(text_group):
+        buckets_per_text.setdefault(g, set()).add(rows[idx]["bucket"])
+    pickups = sum(max(0, len(b) - 1) for b in buckets_per_text.values())
+
+    # Depth: for each newsroom, every distinct piece it filed beyond the first.
+    texts_per_bucket: dict[str, set[int]] = {}
+    for idx, g in enumerate(text_group):
+        texts_per_bucket.setdefault(rows[idx]["bucket"], set()).add(g)
+    depth = sum(max(0, len(t) - 1) for t in texts_per_bucket.values())
+
+    score = origins + pickup_weight * math.log1p(pickups) + depth_weight * math.log1p(depth)
+    return Corroboration(origins, pickups, depth, score)
+
+
+def origin_count(
+    con: sqlite3.Connection,
+    cluster_id: str,
+    syndication_threshold: float = SYNDICATION_THRESHOLD,
+) -> int:
+    """Independent originating newsrooms. See `corroboration` for the full split."""
+    return corroboration(con, cluster_id, syndication_threshold).origins
 
 
 def bucket_count(con: sqlite3.Connection, cluster_id: str) -> int:

news/config.py

@@ -95,6 +95,8 @@ class Config:
             "normalize": s.get("normalize", "zscore-per-block"),
             "veto_absolute": s.get("veto_absolute", 0.60),
             "veto_margin_over_interest": s.get("veto_margin_over_interest", 0.3),
+            "corroboration_pickup_weight": s.get("corroboration_pickup_weight", 0.3),
+            "corroboration_depth_weight": s.get("corroboration_depth_weight", 0.2),
         }
 
     # -- reader ------------------------------------------------------------

profile/config.toml

@@ -57,3 +57,13 @@ normalize = "zscore-per-block"
 # similarity OR when the veto signal beats the best interest signal by a margin.
 veto_absolute = 0.60
 veto_margin_over_interest = 0.3
+
+# Corroboration = origins + pickup_weight*log1p(pickups) + depth_weight*log1p(depth).
+# "Several outlets carried this" covers four different situations; see
+# news/cluster.py:Corroboration. Origins (independent newsrooms doing their own
+# reporting) dominate; the other two break ties among the many two- and
+# three-origin clusters. Measured on a real corpus: 163 origins, 90 depth,
+# 11 pickups -- depth matters far more here because these sources mostly write
+# their own copy. Raise pickup_weight if you add local papers running wire.
+corroboration_pickup_weight = 0.3
+corroboration_depth_weight = 0.2

tests/test_cluster.py

@@ -153,3 +153,114 @@ def test_origin_count_never_exceeds_bucket_count(con):
              ("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)
+
+
+def equicos(n, c=0.90, dim=None):
+    """n unit vectors with pairwise cosine exactly `c`.
+
+    Needed because these tests must sit in a narrow band: above the 0.82
+    clustering threshold so the items form ONE cluster, below the 0.97
+    syndication threshold so they read as distinct pieces of writing. Earlier
+    drafts of this test used ad-hoc vectors that fell outside it in both
+    directions and tested nothing.
+    """
+    dim = dim or n + 1
+    out = []
+    for i in range(n):
+        v = np.zeros(dim, dtype=np.float32)
+        v[0] = np.sqrt(c)
+        v[1 + i] = np.sqrt(1 - c)
+        out.append(v)
+    return out
+
+
+def _mk(con, spec, vec_for):
+    """spec: list of (item_id, source_id, title, content_hash). One cluster."""
+    when = "2026-01-01T00:{:02d}:00+00:00"
+    rows = []
+    for n, (iid, sid, title, chash) in enumerate(spec):
+        con.execute(
+            """INSERT INTO items (id,source_id,title,summary,url,canonical_url,
+                                  published_at,first_seen_at,content_hash)
+               VALUES (?,?,?,'',?,?,?,?,?)""",
+            (iid, sid, title, f"http://x/{iid}", f"http://x/{iid}",
+             when.format(n), when.format(n), chash))
+        v = vec_for(chash)
+        con.execute("INSERT OR REPLACE INTO embeddings (content_hash,dim,vec) VALUES (?,?,?)",
+                    (chash, len(v), store.to_blob(v)))
+    rows = con.execute("SELECT * FROM items ORDER BY published_at").fetchall()
+    v = {r["content_hash"]: vec_for(r["content_hash"]) for r in rows}
+    return cluster.assign(con, list(rows), v)[0].cluster_id
+
+
+def _sources(con, *pairs):
+    for sid, bucket in pairs:
+        store.upsert_source(con, id=sid, name=sid, url=f"http://{sid}", kind="rss",
+                            bucket=bucket, weight=1.0, prominence=0, enabled=1,
+                            poll_interval_secs=900)
+
+
+def test_corroboration_ranks_the_four_cases_in_the_right_order(con):
+    """The 2x2 that motivates the split, each case scored in isolation.
+
+    Ordering must be: four independent newsrooms > four outlets running one
+    wire story > one newsroom filing four pieces > one story on two feeds of
+    the same newsroom.
+    """
+    # "wire" is one shared vector (cosine 1.0 -> syndicated); t0..t3 and u0..u3
+    # are pairwise 0.90 -> one cluster, four distinct pieces of writing.
+    distinct = equicos(4, 0.90, dim=6)
+    table = {"wire": np.eye(6, dtype=np.float32)[0]}
+    for i in range(4):
+        table[f"t{i}"] = distinct[i]
+        table[f"u{i}"] = distinct[i]
+
+    def vec(h):
+        return table[h]
+
+    # (a) one newsroom, one story, two feeds -> pure duplicate
+    c1 = store.connect(":memory:")
+    _sources(c1, ("f1", "bbc"), ("f2", "bbc"))
+    cid = _mk(c1, [("a1", "f1", "Same headline", "wire"),
+                   ("a2", "f2", "Same headline", "wire")], vec)
+    dup = cluster.corroboration(c1, cid)
+
+    # (b) one newsroom, four distinct pieces -> depth
+    c2 = store.connect(":memory:")
+    _sources(c2, ("g1", "nyt"))
+    cid = _mk(c2, [(f"b{i}", "g1", f"Distinct angle {i}", f"t{i}") for i in range(4)], vec)
+    deep = cluster.corroboration(c2, cid)
+
+    # (c) four newsrooms, one syndicated wire story -> pickups
+    c3 = store.connect(":memory:")
+    _sources(c3, *[(f"h{i}", f"local{i}") for i in range(4)])
+    cid = _mk(c3, [(f"c{i}", f"h{i}", "Wire dispatch text", "wire") for i in range(4)], vec)
+    wire = cluster.corroboration(c3, cid)
+
+    # (d) four newsrooms each writing their own -> origins
+    c4 = store.connect(":memory:")
+    _sources(c4, *[(f"k{i}", f"nat{i}") for i in range(4)])
+    cid = _mk(c4, [(f"d{i}", f"k{i}", f"Own reporting {i}", f"u{i}") for i in range(4)], vec)
+    orig = cluster.corroboration(c4, cid)
+
+    assert dup.origins == 1 and dup.pickups == 0 and dup.depth == 0
+    assert deep.origins == 1 and deep.depth == 3
+    assert wire.origins == 1 and wire.pickups == 3
+    assert orig.origins == 4
+
+    assert orig.score > wire.score > deep.score > dup.score, (
+        f"expected origins > pickups > depth > duplicate, got "
+        f"{orig.score:.2f} / {wire.score:.2f} / {deep.score:.2f} / {dup.score:.2f}")
+
+
+def test_origins_dominate_the_tiebreakers(con):
+    """No amount of depth or pickups may outweigh one extra real newsroom."""
+    c = store.connect(":memory:")
+    _sources(c, ("m1", "one"))
+    vs = equicos(20, 0.90)
+    table = {f"v{i}": vs[i] for i in range(20)}
+    cid = _mk(c, [(f"e{i}", "m1", f"Piece {i}", f"v{i}") for i in range(20)],
+              lambda h: table[h])
+    twenty_deep = cluster.corroboration(c, cid)
+    assert twenty_deep.score < 2.0, (
+        "20 pieces from one newsroom must still rank below two independent ones")