connorandClaude Opus 5
1e06d1a2b0
Clustering runs at assignment time only, so tightening the rules changed what happened to new items and left every stored cluster exactly as it was. Sixteen blobs holding 2,379 items stayed frozen in the database after the fix shipped, including the 110-item "Losing control of AI is the plan" cluster a triage card was still reporting as 110 sources. `news recluster` re-applies the current rules to stored items, and preserves cluster IDs while doing it: each cluster keeps its row and its founding item, and only the other members are detached and re-assigned. Rebuilding from scratch would mint new ids and orphan all 246 rows in notification_log, digest_items and feedback -- blanking every past digest page, which is the thing that was just fixed. A test pins that property. Run against production, against a backup at data/news2/news.db.pre-recluster: largest cluster 665 -> 39, clusters of 50+ 16 -> 0, multi-item clusters 1,964 -> 2,032, every foreign reference still resolving (196/196, 46/46, 4/4), and no item in two clusters. A following poll cycle formed no new blob. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015xPLgmcXCc9JjVwmQF63eM
445 lines
21 KiB
Python
445 lines
21 KiB
Python
"""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"
|
|
|
|
|
|
def test_syndicated_partners_in_one_bucket_count_once(con):
|
|
"""Regression: Ohio Capital Journal and Kentucky Lantern are both States
|
|
Newsroom and carried 10 identical headlines in one poll. Configured as
|
|
separate buckets they would have doubled the corroboration signal."""
|
|
v = unit(1, 0, 0)
|
|
for sid, name in (("sn1", "Ohio Capital Journal"), ("sn2", "Kentucky Lantern")):
|
|
store.upsert_source(con, id=sid, name=name, url=f"http://{sid}", kind="rss",
|
|
bucket="states-newsroom", weight=1.0, prominence=0,
|
|
enabled=1, poll_interval_secs=900)
|
|
rows = [add(con, "s1i", "sn1", "2026-01-01T00:00:00+00:00", "hs1"),
|
|
add(con, "s2i", "sn2", "2026-01-01T00:30:00+00:00", "hs2")]
|
|
out = cluster.assign(con, rows, {"hs1": v, "hs2": v})
|
|
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)
|
|
|
|
|
|
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")
|
|
|
|
|
|
def _at(degrees):
|
|
"""A unit vector at a given angle from the x axis, in the x-y plane."""
|
|
import math
|
|
rad = math.radians(degrees)
|
|
return unit(math.cos(rad), math.sin(rad), 0)
|
|
|
|
|
|
def test_the_centroid_cannot_drift_away_from_the_founding_story(con):
|
|
"""Regression: merging on the running mean alone let a cluster walk.
|
|
|
|
Measured on the live corpus: a 110-item cluster held only 10 members within
|
|
0.82 of its founding story, while all 110 were within 0.82 of the drifted
|
|
centroid. Each merge pulls the mean toward what has accumulated, so the mean
|
|
tracks the chain and the cluster ends up matching a whole subject rather
|
|
than one event.
|
|
|
|
Twenty items five degrees apart: neighbours are cosine 0.996, but the last
|
|
is 95 degrees from the first. A mean-only rule swallows nearly all of them.
|
|
"""
|
|
vecs, rows = {}, []
|
|
for n in range(20):
|
|
h = f"h{n}"
|
|
vecs[h] = _at(n * 5)
|
|
rows.append(add(con, f"i{n:02d}", "s1",
|
|
f"2026-01-0{n // 8 + 1}T{n % 8 * 3:02d}:00:00+00:00", h))
|
|
|
|
def biggest(**kw):
|
|
con.execute("DELETE FROM cluster_members")
|
|
con.execute("DELETE FROM clusters")
|
|
out = cluster.assign(con, rows, vecs, threshold=0.82, **kw)
|
|
sizes = {}
|
|
for a in out:
|
|
sizes[a.cluster_id] = sizes.get(a.cluster_id, 0) + 1
|
|
return max(sizes.values()), len(sizes)
|
|
|
|
drifted_max, drifted_n = biggest(anchor_threshold=0.0)
|
|
anchored_max, anchored_n = biggest(anchor_threshold=0.80)
|
|
|
|
# The mean walks the chain and swallows most of a 95-degree arc.
|
|
assert drifted_max >= 12, (
|
|
f"expected the mean-only rule to chain; largest was only {drifted_max}")
|
|
# cos(37 deg) ~ 0.80, so a founding cluster spans about seven 5-degree steps.
|
|
assert anchored_max <= 9, (
|
|
f"the anchor let {anchored_max} items into one cluster; it should cap near 8")
|
|
assert anchored_n > drifted_n
|
|
|
|
|
|
def _disc_sources(con):
|
|
store.upsert_source(con, id="hn", name="HN", url="http://hn", kind="hn",
|
|
bucket="hacker-news", weight=1.0, prominence=0, enabled=1,
|
|
poll_interval_secs=900)
|
|
store.upsert_source(con, id="lob", name="Lob", url="http://lob", kind="rss",
|
|
bucket="lobsters", weight=1.0, prominence=0, enabled=1,
|
|
poll_interval_secs=900)
|
|
|
|
|
|
def test_two_discussion_posts_on_one_topic_are_not_one_story(con):
|
|
"""HN and Lobsters publish posts ABOUT a subject, not reports OF an event.
|
|
|
|
Counting them as one cluster made them corroborate each other, which is how
|
|
a 'Losing control of AI' cluster reached 110 members across 8 days.
|
|
Measured separation: real cross-posts sit at 0.990 similarity, topical
|
|
merges at 0.769.
|
|
"""
|
|
_disc_sources(con)
|
|
topical_a, topical_b = unit(1, 0.0, 0), unit(1, 0.45, 0) # ~0.91...
|
|
rows = [add(con, "d1", "hn", "2026-01-01T00:00:00+00:00", "ha"),
|
|
add(con, "d2", "lob", "2026-01-01T01:00:00+00:00", "hb")]
|
|
# cosine here is ~0.912: above the topical bar, below the discussion bar.
|
|
out = cluster.assign(con, rows, {"ha": topical_a, "hb": topical_b},
|
|
threshold=0.82, discussion_threshold=0.95,
|
|
anchor_threshold=0.0)
|
|
assert len(set(a.cluster_id for a in out)) == 2, (
|
|
"two discussion posts below the near-duplicate bar must stay separate")
|
|
|
|
|
|
def test_the_same_link_cross_posted_still_merges(con):
|
|
"""The one discussion merge worth making: one article, two aggregators."""
|
|
_disc_sources(con)
|
|
v = unit(1, 0.01, 0)
|
|
rows = [add(con, "d1", "hn", "2026-01-01T00:00:00+00:00", "ha"),
|
|
add(con, "d2", "lob", "2026-01-01T01:00:00+00:00", "hb")]
|
|
out = cluster.assign(con, rows, {"ha": v, "hb": unit(1, 0.02, 0)},
|
|
threshold=0.82, discussion_threshold=0.90)
|
|
assert len(set(a.cluster_id for a in out)) == 1
|
|
|
|
|
|
def test_a_newsroom_report_and_a_discussion_post_still_merge(con):
|
|
"""A mixed cluster is one story: the near-duplicate bar applies only when
|
|
BOTH sides are discussion."""
|
|
_disc_sources(con)
|
|
a, b = unit(1, 0.0, 0), unit(1, 0.45, 0) # ~0.91, below discussion bar
|
|
rows = [add(con, "n1", "s1", "2026-01-01T00:00:00+00:00", "ha"),
|
|
add(con, "d1", "hn", "2026-01-01T01:00:00+00:00", "hb")]
|
|
out = cluster.assign(con, rows, {"ha": a, "hb": b},
|
|
threshold=0.82, discussion_threshold=0.95,
|
|
anchor_threshold=0.0)
|
|
assert len(set(a.cluster_id for a in out)) == 1, (
|
|
"newsroom + discussion on one event must still merge at the topical bar")
|
|
|
|
|
|
def test_lowering_the_threshold_is_not_silently_overridden_by_the_anchor(con):
|
|
"""A knob that does nothing is worse than no knob.
|
|
|
|
The anchor is clamped to `threshold`, so loosening clustering in config
|
|
actually loosens it instead of running into a second, invisible bar.
|
|
"""
|
|
a, b = unit(1, 0, 0), unit(0.6, 0.8, 0) # cosine 0.6, below the 0.80 default
|
|
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")]
|
|
out = cluster.assign(con, rows, {"h1": a, "h2": b}, threshold=0.5)
|
|
assert len(set(x.cluster_id for x in out)) == 1, (
|
|
"threshold=0.5 must merge a 0.6 pair; the anchor should have clamped to 0.5")
|
|
|
|
|
|
def test_recluster_preserves_cluster_ids(tmp_path, monkeypatch, capsys):
|
|
"""Clustering rules run only at assignment time, so tightening one leaves
|
|
every existing cluster untouched. `news recluster` re-applies them.
|
|
|
|
The property that matters: cluster IDs SURVIVE. notification_log,
|
|
digest_items and feedback all reference cluster_id, so minting new ids would
|
|
orphan every past digest page. Each cluster keeps its row and its founding
|
|
item; only the other members move.
|
|
"""
|
|
from news.cli import main
|
|
|
|
db = tmp_path / "rc.db"
|
|
monkeypatch.setenv("NEWS_DB", str(db))
|
|
con = store.connect(db)
|
|
store.upsert_source(con, id="s1", name="A", url="http://a", kind="rss", bucket="a",
|
|
weight=1.0, prominence=0, enabled=1, poll_interval_secs=900)
|
|
|
|
# A blob: a founding item plus four that a 0.80 anchor would never admit.
|
|
import math
|
|
vecs = {}
|
|
for n in range(5):
|
|
h = f"h{n}"
|
|
rad = math.radians(n * 20) # 0, 20, 40, 60, 80 degrees from the lead
|
|
vecs[h] = unit(math.cos(rad), math.sin(rad), 0)
|
|
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", f"story {n}", f"http://x/{n}", f"http://x/{n}",
|
|
f"2026-01-01T0{n}:00:00+00:00", f"2026-01-01T0{n}:00:00+00:00", h))
|
|
con.execute("INSERT INTO embeddings (content_hash,dim,vec) VALUES (?,?,?)",
|
|
(h, 3, store.to_blob(vecs[h])))
|
|
con.execute("""INSERT INTO clusters (id,centroid,size,first_seen_at,last_seen_at,
|
|
lead_item_id) VALUES ('blob',?,5,?,?,'i0')""",
|
|
(store.to_blob(vecs["h0"]), "2026-01-01T00:00:00+00:00",
|
|
"2026-01-01T04:00:00+00:00"))
|
|
for n in range(5):
|
|
con.execute("INSERT INTO cluster_members VALUES ('blob',?,?)",
|
|
(f"i{n}", "2026-01-01T00:00:00+00:00"))
|
|
con.execute("""INSERT INTO digest_items (day,cluster_id,rank,score,reason)
|
|
VALUES ('2026-01-01','blob',0,1.0,'digest')""")
|
|
con.execute("""INSERT INTO notification_log (cluster_id,lane,score,created_at)
|
|
VALUES ('blob','alert:interest',1.0,'2026-01-01T00:00:00+00:00')""")
|
|
|
|
assert main(["recluster"]) == 0
|
|
|
|
con = store.connect(db)
|
|
assert con.execute("SELECT COUNT(*) n FROM clusters WHERE id='blob'").fetchone()["n"] == 1, \
|
|
"the cluster id must survive, or every digest_items row is orphaned"
|
|
assert con.execute(
|
|
"SELECT lead_item_id FROM clusters WHERE id='blob'").fetchone()["lead_item_id"] == "i0"
|
|
size = con.execute("SELECT size FROM clusters WHERE id='blob'").fetchone()["size"]
|
|
assert size < 5, f"the blob still holds {size} members; the anchor should have split it"
|
|
# Every foreign reference still resolves.
|
|
for table in ("digest_items", "notification_log"):
|
|
assert con.execute(
|
|
f"SELECT COUNT(*) n FROM {table} t JOIN clusters c ON c.id = t.cluster_id"
|
|
).fetchone()["n"] == 1, f"{table} reference was orphaned"
|
|
# Detached members are not lost -- every item is in exactly one cluster.
|
|
assert con.execute("SELECT COUNT(*) n FROM cluster_members").fetchone()["n"] == 5
|
|
assert con.execute(
|
|
"SELECT COUNT(*) n FROM (SELECT item_id FROM cluster_members GROUP BY item_id "
|
|
"HAVING COUNT(*) > 1)").fetchone()["n"] == 0, "an item ended up in two clusters"
|