Add news recluster, and apply the new rules to stored clusters
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
This commit is contained in:
1 parent
835a7ab538
commit
1e06d1a2b0
2 files changed
+166
No files matched your search
+103
@@ -281,6 +281,102 @@ def cmd_digest(args) -> int:
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_recluster(args) -> int:
|
||||
"""Re-apply the current clustering rules to items already stored.
|
||||
|
||||
Clustering runs at assignment time only, so changing a threshold leaves every
|
||||
existing cluster exactly as it was. After tightening the rules on 2026-09-17
|
||||
the new ones held, but 16 blobs holding 2,379 items stayed frozen in the
|
||||
database -- including a 110-item "Losing control of AI" cluster that a triage
|
||||
card still reported as 110 sources.
|
||||
|
||||
Cluster IDs are PRESERVED. Each cluster keeps its row and its founding item;
|
||||
only the other members are detached and re-assigned. Rebuilding from scratch
|
||||
would mint new ids and orphan every row in notification_log, digest_items and
|
||||
feedback -- 246 of them -- blanking every past digest page.
|
||||
"""
|
||||
from datetime import UTC, datetime
|
||||
|
||||
cfg = Config()
|
||||
con = store.connect(cfg.db_path)
|
||||
c = cfg.clustering
|
||||
|
||||
targets = con.execute(
|
||||
"SELECT id, size, lead_item_id FROM clusters WHERE size >= ? ORDER BY size DESC",
|
||||
(max(args.min_size, 2),),
|
||||
).fetchall()
|
||||
held = sum(r["size"] for r in targets)
|
||||
print(f"{len(targets)} clusters of {args.min_size}+ holding {held} items")
|
||||
if not targets:
|
||||
return 0
|
||||
|
||||
if args.dry_run:
|
||||
for r in targets[:15]:
|
||||
lead = con.execute("SELECT title FROM items WHERE id=?",
|
||||
(r["lead_item_id"],)).fetchone()
|
||||
print(f" {r['size']:>4} {(lead['title'] if lead else '?')[:62]}")
|
||||
print(f"\n[dry-run] would detach {held - len(targets)} members and re-assign "
|
||||
f"them under threshold={c['threshold']} anchor={c['anchor_threshold']} "
|
||||
f"discussion={c['discussion_threshold']}")
|
||||
return 0
|
||||
|
||||
# Detach everything but the founding item, and rewind the cluster to it.
|
||||
orphan_ids: list[str] = []
|
||||
for r in targets:
|
||||
members = con.execute(
|
||||
"""SELECT cm.item_id, i.published_at, i.content_hash
|
||||
FROM cluster_members cm JOIN items i ON i.id = cm.item_id
|
||||
WHERE cm.cluster_id = ? ORDER BY i.published_at""", (r["id"],)).fetchall()
|
||||
if not members:
|
||||
continue
|
||||
keep = r["lead_item_id"] or members[0]["item_id"]
|
||||
lead_row = next((m for m in members if m["item_id"] == keep), members[0])
|
||||
vec = store.cached_vectors(con, [lead_row["content_hash"]]).get(
|
||||
lead_row["content_hash"])
|
||||
for m in members:
|
||||
if m["item_id"] == lead_row["item_id"]:
|
||||
continue
|
||||
con.execute("DELETE FROM cluster_members WHERE cluster_id=? AND item_id=?",
|
||||
(r["id"], m["item_id"]))
|
||||
orphan_ids.append(m["item_id"])
|
||||
con.execute(
|
||||
"""UPDATE clusters SET size=1, last_seen_at=?, first_seen_at=?,
|
||||
lead_item_id=?""" +
|
||||
(", centroid=?" if vec is not None else "") + " WHERE id=?",
|
||||
((lead_row["published_at"], lead_row["published_at"], lead_row["item_id"])
|
||||
+ ((store.to_blob(vec),) if vec is not None else ()) + (r["id"],)))
|
||||
|
||||
print(f"detached {len(orphan_ids)} members; re-assigning")
|
||||
rows, vectors = [], {}
|
||||
for i in range(0, len(orphan_ids), 500):
|
||||
chunk = orphan_ids[i : i + 500]
|
||||
marks = ",".join("?" * len(chunk))
|
||||
rows.extend(con.execute(f"SELECT * FROM items WHERE id IN ({marks})", chunk))
|
||||
vectors = store.cached_vectors(con, [r["content_hash"] for r in rows])
|
||||
rows = [r for r in rows if r["content_hash"] in vectors]
|
||||
|
||||
started = datetime.now(UTC)
|
||||
cluster_mod.assign(con, rows, vectors,
|
||||
threshold=c["threshold"], window_hours=c["window_hours"],
|
||||
anchor_threshold=c["anchor_threshold"],
|
||||
discussion_threshold=c["discussion_threshold"])
|
||||
# `assign` keeps sizes for clusters it touches, but a cluster rewound to its
|
||||
# founder and never rejoined still claims whatever it was reset to.
|
||||
con.execute("""UPDATE clusters SET size = (
|
||||
SELECT COUNT(*) FROM cluster_members cm WHERE cm.cluster_id = clusters.id)""")
|
||||
con.execute("""UPDATE clusters SET last_seen_at = COALESCE((
|
||||
SELECT MAX(i.published_at) FROM cluster_members cm
|
||||
JOIN items i ON i.id = cm.item_id
|
||||
WHERE cm.cluster_id = clusters.id), last_seen_at)""")
|
||||
con.execute("DELETE FROM clusters WHERE size = 0")
|
||||
|
||||
big = con.execute("SELECT COUNT(*) n FROM clusters WHERE size >= 50").fetchone()["n"]
|
||||
mx = con.execute("SELECT MAX(size) m FROM clusters").fetchone()["m"]
|
||||
print(f"done in {(datetime.now(UTC)-started).total_seconds():.0f}s: "
|
||||
f"largest cluster now {mx}, clusters of 50+: {big}")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_serve(args) -> int:
|
||||
import uvicorn
|
||||
|
||||
@@ -321,6 +417,13 @@ def main(argv: list[str] | None = None) -> int:
|
||||
help="rebuild the digest and its prose without sending ntfy")
|
||||
p.set_defaults(fn=cmd_digest)
|
||||
|
||||
p = sub.add_parser("recluster",
|
||||
help="re-apply current clustering rules to stored items")
|
||||
p.add_argument("--min-size", type=int, default=2,
|
||||
help="only rebuild clusters at least this large (default 2)")
|
||||
p.add_argument("--dry-run", action="store_true")
|
||||
p.set_defaults(fn=cmd_recluster)
|
||||
|
||||
p = sub.add_parser("serve", help="run the web server")
|
||||
p.add_argument("--host", default="0.0.0.0")
|
||||
p.add_argument("--port", type=int, default=3000)
|
||||
|
||||
@@ -379,3 +379,66 @@ def test_lowering_the_threshold_is_not_silently_overridden_by_the_anchor(con):
|
||||
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"
|
||||
Reference in new issue
Block a user