diff --git a/.gitignore b/.gitignore
index 122f031..cb3e9a1 100644
--- a/.gitignore
+++ b/.gitignore
@@ -30,3 +30,6 @@ tests/**/*.pyc
**/*.db-shm
**/*.npy
.venv/**
+
+# Local runtime data
+data/**
diff --git a/news/__init__.py b/news/__init__.py
new file mode 100644
index 0000000..bac9617
--- /dev/null
+++ b/news/__init__.py
@@ -0,0 +1,3 @@
+"""Personal news aggregator: many sources in, a few well-chosen notifications out."""
+
+__version__ = "0.1.0"
diff --git a/news/cli.py b/news/cli.py
new file mode 100644
index 0000000..c09caa4
--- /dev/null
+++ b/news/cli.py
@@ -0,0 +1,82 @@
+"""Command line entry point."""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from . import cluster as cluster_mod
+from . import store
+from .config import Config
+from .embed import Embedder
+from .pipeline import run_cycle
+
+
+def cmd_poll(args) -> int:
+ cfg = Config()
+ con = store.connect(cfg.db_path)
+ report = run_cycle(con, cfg, Embedder())
+ print(report)
+ for err in report.errors or []:
+ print(f" ! {err}", file=sys.stderr)
+ return 0
+
+
+def cmd_sources(args) -> int:
+ cfg = Config()
+ con = store.connect(cfg.db_path)
+ rows = con.execute(
+ """SELECT s.name, s.kind, s.bucket, s.enabled, s.last_status,
+ COUNT(i.id) AS items
+ FROM sources s LEFT JOIN items i ON i.source_id = s.id
+ GROUP BY s.id ORDER BY items DESC"""
+ ).fetchall()
+ print(f"{'source':26} {'kind':7} {'bucket':22} {'items':>6} status")
+ for r in rows:
+ flag = " " if r["enabled"] else "-"
+ print(f"{flag}{r['name'][:25]:25} {r['kind']:7} {r['bucket'][:21]:22} "
+ f"{r['items']:>6} {(r['last_status'] or '')[:40]}")
+ return 0
+
+
+def cmd_clusters(args) -> int:
+ cfg = Config()
+ con = store.connect(cfg.db_path)
+ rows = con.execute(
+ """SELECT c.id, c.size, c.first_seen_at,
+ (SELECT title FROM items WHERE id = c.lead_item_id) AS lead
+ FROM clusters c WHERE c.size >= ?
+ ORDER BY c.size DESC, c.last_seen_at DESC LIMIT ?""",
+ (args.min_size, args.limit),
+ ).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']}")
+ if args.verbose:
+ for m in con.execute(
+ """SELECT i.title, s.name 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=? LIMIT 8""", (r["id"],)):
+ print(f" [{m['name'][:16]:16}] {m['title'][:70]}")
+ return 0
+
+
+def main(argv: list[str] | None = None) -> int:
+ parser = argparse.ArgumentParser(prog="news")
+ sub = parser.add_subparsers(dest="cmd", required=True)
+
+ sub.add_parser("poll", help="run one fetch/embed/cluster cycle").set_defaults(fn=cmd_poll)
+ sub.add_parser("sources", help="list sources and item counts").set_defaults(fn=cmd_sources)
+
+ p = sub.add_parser("clusters", help="inspect clusters")
+ p.add_argument("--min-size", type=int, default=2)
+ p.add_argument("--limit", type=int, default=20)
+ p.add_argument("-v", "--verbose", action="store_true")
+ p.set_defaults(fn=cmd_clusters)
+
+ args = parser.parse_args(argv)
+ return args.fn(args)
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/news/cluster.py b/news/cluster.py
new file mode 100644
index 0000000..df51d89
--- /dev/null
+++ b/news/cluster.py
@@ -0,0 +1,149 @@
+"""Incremental story clustering by embedding similarity.
+
+Replaces the old system's Jaccard over title tokens, which could not see
+paraphrase: "Gloria Steinem" coverage scored 0.33 against a 0.5 threshold and
+every story stayed a singleton, so corroboration -- and therefore the whole
+importance axis -- read 1 for everything.
+
+THRESHOLD is calibrated, not guessed. On 3,885 real items:
+
+ 0.70 over-merges badly ("Solheim Cup capsules" joins "FIBA Women's World Cup
+ Glance"; two different baseball games merge)
+ 0.74 still mixes distinct stories that share a topic
+ 0.82 chosen. 35% of items land in multi-item clusters and the large ones are
+ clean: 76 items on the AfD Saxony-Anhalt election, 28 on the Trump
+ envoys to Moscow and Kyiv, 27 on the Nepal tunnel rescue, 22 on the UK
+ West Bank settlement sanctions.
+ 0.86 under-merges; only 22% of items cluster.
+
+Over-merging is the worse failure: it inflates the corroboration signal that
+feeds importance, and it merges unrelated stories into one notification.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+
+import numpy as np
+
+from . import store
+from .util import stable_id
+
+THRESHOLD = 0.82
+WINDOW_HOURS = 48
+
+
+@dataclass(frozen=True)
+class Assignment:
+ item_id: str
+ cluster_id: str
+ created: bool
+ similarity: float
+
+
+def _parse(ts: str) -> datetime:
+ return datetime.fromisoformat(ts)
+
+
+def assign(
+ con: sqlite3.Connection,
+ items: list[sqlite3.Row],
+ vectors: dict[str, np.ndarray],
+ threshold: float = THRESHOLD,
+ window_hours: int = WINDOW_HOURS,
+) -> list[Assignment]:
+ """Assign items to clusters, oldest first.
+
+ Order matters: processing in publication order makes the result stable and
+ reproducible in replay, which is what makes threshold tuning meaningful.
+ """
+ out: list[Assignment] = []
+ ordered = sorted(items, key=lambda r: r["published_at"])
+
+ for item in ordered:
+ vec = vectors.get(item["content_hash"])
+ if vec is None:
+ continue
+ # Guard against double-counting: re-assigning an item that already has a
+ # cluster would increment that cluster's size again and corrupt the
+ # corroboration signal. The pipeline only passes unclustered items, but
+ # replay and manual invocation must be safe too.
+ existing = con.execute(
+ "SELECT cluster_id FROM cluster_members WHERE item_id=?", (item["id"],)
+ ).fetchone()
+ if existing is not None:
+ out.append(Assignment(item["id"], existing["cluster_id"], False, 1.0))
+ continue
+ published = _parse(item["published_at"])
+ cutoff = (published - timedelta(hours=window_hours)).isoformat()
+
+ # Only clusters still warm can absorb the item. A 2h grace on the upper
+ # bound lets a late-arriving report join a story it predates.
+ rows = con.execute(
+ """SELECT id, centroid, size FROM clusters
+ WHERE last_seen_at >= ? AND first_seen_at <= ?""",
+ (cutoff, (published + timedelta(hours=2)).isoformat()),
+ ).fetchall()
+
+ best_id, best_sim = None, -1.0
+ if rows:
+ mat = np.stack([store.from_blob(r["centroid"]) for r in rows])
+ mat /= np.maximum(np.linalg.norm(mat, axis=1, keepdims=True), 1e-12)
+ sims = mat @ vec
+ k = int(np.argmax(sims))
+ best_id, best_sim = rows[k]["id"], float(sims[k])
+
+ if best_id is not None and best_sim >= threshold:
+ row = next(r for r in rows if r["id"] == best_id)
+ size = row["size"]
+ centroid = (store.from_blob(row["centroid"]) * size + vec) / (size + 1)
+ con.execute(
+ """UPDATE clusters SET centroid=?, size=?, last_seen_at=max(last_seen_at,?)
+ WHERE id=?""",
+ (store.to_blob(centroid), size + 1, item["published_at"], best_id),
+ )
+ cluster_id, created = best_id, False
+ else:
+ cluster_id = stable_id("cluster", item["id"])
+ con.execute(
+ """INSERT OR IGNORE INTO clusters
+ (id,centroid,size,first_seen_at,last_seen_at,lead_item_id)
+ VALUES (?,?,?,?,?,?)""",
+ (
+ cluster_id,
+ store.to_blob(vec),
+ 1,
+ item["published_at"],
+ item["published_at"],
+ item["id"],
+ ),
+ )
+ created = True
+
+ con.execute(
+ """INSERT OR IGNORE INTO cluster_members (cluster_id,item_id,added_at)
+ VALUES (?,?,datetime('now'))""",
+ (cluster_id, item["id"]),
+ )
+ out.append(Assignment(item["id"], cluster_id, created, best_sim))
+ return out
+
+
+def bucket_count(con: sqlite3.Connection, cluster_id: str) -> int:
+ """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
+ and then fired a priority-5 push on it.
+ """
+ row = con.execute(
+ """SELECT COUNT(DISTINCT s.bucket) AS n
+ 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,),
+ ).fetchone()
+ return int(row["n"] or 0)
diff --git a/news/config.py b/news/config.py
new file mode 100644
index 0000000..45813bc
--- /dev/null
+++ b/news/config.py
@@ -0,0 +1,46 @@
+"""Configuration loading: sources, profile, and runtime paths."""
+
+from __future__ import annotations
+
+import os
+import tomllib
+from dataclasses import dataclass, field
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent
+
+
+def _path(env: str, default: Path) -> Path:
+ return Path(os.environ.get(env, default))
+
+
+@dataclass
+class Config:
+ db_path: Path = field(default_factory=lambda: _path("NEWS_DB", REPO / "data" / "news.db"))
+ sources_path: Path = field(
+ default_factory=lambda: _path("NEWS_SOURCES", REPO / "profile" / "sources.toml")
+ )
+ profile_path: Path = field(
+ default_factory=lambda: _path("NEWS_PROFILE", REPO / "profile" / "profile.toml")
+ )
+ user_agent: str = (
+ "news/0.1 (+https://news.rcjohnstone.com; personal feed reader; one user)"
+ )
+
+ def sources(self) -> list[dict]:
+ raw = tomllib.loads(self.sources_path.read_text())
+ defaults = raw.get("defaults", {})
+ out = []
+ for entry in raw.get("source", []):
+ merged = {**defaults, **entry}
+ merged.setdefault("kind", "rss")
+ merged.setdefault("bucket", merged["name"].lower().replace(" ", "-"))
+ merged.setdefault("weight", 1.0)
+ merged.setdefault("prominence", 0)
+ merged.setdefault("enabled", True)
+ merged.setdefault("poll_interval_secs", 900)
+ out.append(merged)
+ return out
+
+ def profile(self) -> dict:
+ return tomllib.loads(self.profile_path.read_text())
diff --git a/news/embed.py b/news/embed.py
new file mode 100644
index 0000000..1d313d9
--- /dev/null
+++ b/news/embed.py
@@ -0,0 +1,64 @@
+"""Embedding, on CPU, with a content-hash cache.
+
+Deliberately fastembed (quantized ONNX) rather than sentence-transformers: no
+torch, no CUDA, ~130 MB of model. It never touches the GPU, so it cannot
+contend with the llama-swap inference stack.
+
+Measured on this box: ~11 items/sec with default settings. That is far slower
+than the "thousands/sec" the design originally assumed, but irrelevant in
+practice -- a ~2,000-item day embeds in about three minutes, and the cache means
+each distinct text is embedded exactly once.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+from collections.abc import Sequence
+
+import numpy as np
+
+from . import store
+
+MODEL = "BAAI/bge-small-en-v1.5"
+DIM = 384
+
+
+class Embedder:
+ def __init__(self, model_name: str = MODEL) -> None:
+ self._model_name = model_name
+ self._model = None # loaded lazily: importing fastembed is slow
+
+ def _load(self):
+ if self._model is None:
+ from fastembed import TextEmbedding
+
+ self._model = TextEmbedding(self._model_name)
+ return self._model
+
+ def encode(self, texts: Sequence[str]) -> np.ndarray:
+ """L2-normalised embeddings, so a dot product is cosine similarity."""
+ if not texts:
+ return np.zeros((0, DIM), dtype=np.float32)
+ vecs = np.array(list(self._load().embed(list(texts))), dtype=np.float32)
+ norms = np.linalg.norm(vecs, axis=1, keepdims=True)
+ return vecs / np.maximum(norms, 1e-12)
+
+ def encode_cached(
+ self, con: sqlite3.Connection, pairs: Sequence[tuple[str, str]]
+ ) -> dict[str, np.ndarray]:
+ """`pairs` is (content_hash, text). Only cache misses are embedded."""
+ hashes = [h for h, _ in pairs]
+ have = store.cached_vectors(con, hashes)
+ todo = {h: t for h, t in pairs if h not in have}
+ if todo:
+ keys = list(todo)
+ fresh = self.encode([todo[k] for k in keys])
+ store.store_vectors(con, zip(keys, fresh))
+ have.update(dict(zip(keys, fresh)))
+ return have
+
+
+def embed_text(title: str, summary: str = "", limit: int = 400) -> str:
+ """The canonical text form. Must match everywhere or the cache is useless."""
+ summary = " ".join((summary or "").split())[:limit]
+ return f"{title.strip()}. {summary}".strip()
diff --git a/news/pipeline.py b/news/pipeline.py
new file mode 100644
index 0000000..847e8b6
--- /dev/null
+++ b/news/pipeline.py
@@ -0,0 +1,129 @@
+"""One poll cycle: fetch -> store -> embed -> cluster.
+
+Kept separate from the CLI and the (future) scheduler so replay and the live
+service run exactly the same code. Divergence between the tuning path and the
+production path is what made the old system's replay harness useless.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import sqlite3
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+
+import httpx
+
+from . import cluster, store
+from .config import Config
+from .embed import Embedder, embed_text
+from .sources import fetch
+from .util import stable_id
+
+
+@dataclass
+class CycleReport:
+ fetched: int = 0
+ new_items: int = 0
+ embedded: int = 0
+ clustered: int = 0
+ new_clusters: int = 0
+ errors: list[str] = field(default_factory=list)
+ rate_limited: list[str] = field(default_factory=list)
+
+ def __str__(self) -> str:
+ errs = f", {len(self.errors)} errors" if self.errors else ""
+ if self.rate_limited:
+ errs += f", rate-limited: {', '.join(self.rate_limited)}"
+ return (
+ f"fetched {self.fetched}, new {self.new_items}, embedded {self.embedded}, "
+ f"clustered {self.clustered} ({self.new_clusters} new){errs}"
+ )
+
+
+def sync_sources(con: sqlite3.Connection, cfg: Config) -> list[dict]:
+ """Config is authoritative. Returns the enabled sources with their ids."""
+ out = []
+ for src in cfg.sources():
+ src = dict(src)
+ src["id"] = stable_id("source", src["url"])
+ src["enabled"] = 1 if src.get("enabled", True) else 0
+ store.upsert_source(con, **{k: src[k] for k in (
+ "id", "name", "url", "kind", "bucket", "weight",
+ "prominence", "enabled", "poll_interval_secs")})
+ if src["enabled"]:
+ row = con.execute(
+ "SELECT etag, last_modified FROM sources WHERE id=?", (src["id"],)
+ ).fetchone()
+ src["etag"], src["last_modified"] = row["etag"], row["last_modified"]
+ out.append(src)
+ return out
+
+
+async def fetch_all(cfg: Config, sources: list[dict], concurrency: int = 8):
+ """Fetch every source concurrently. A failing source never blocks the others."""
+ results: list[tuple[dict, list[dict], dict, str | None]] = []
+ sem = asyncio.Semaphore(concurrency)
+
+ async with httpx.AsyncClient(
+ follow_redirects=True,
+ timeout=httpx.Timeout(30.0, connect=10.0),
+ headers={"User-Agent": cfg.user_agent, "Accept": "*/*"},
+ ) as client:
+
+ async def one(src):
+ async with sem:
+ try:
+ items, state = await fetch(client, src)
+ return src, items, state, None
+ except Exception as exc: # noqa: BLE001 - one bad feed must not stop a cycle
+ return src, [], {}, f"{src['name']}: {type(exc).__name__}: {exc}"
+
+ results = await asyncio.gather(*(one(s) for s in sources))
+ return results
+
+
+def run_cycle(con: sqlite3.Connection, cfg: Config, embedder: Embedder) -> CycleReport:
+ report = CycleReport()
+ sources = sync_sources(con, cfg)
+ results = asyncio.run(fetch_all(cfg, sources))
+
+ for src, items, state, err in results:
+ if err:
+ report.errors.append(err)
+ con.execute(
+ "UPDATE sources SET last_polled_at=?, last_status=? WHERE id=?",
+ (datetime.now(UTC).isoformat(), err[:200], src["id"]),
+ )
+ continue
+ if state.get("rate_limited"):
+ report.rate_limited.append(src["name"])
+ report.fetched += len(items)
+ if items:
+ report.new_items += store.insert_items(con, items)
+ con.execute(
+ """UPDATE sources SET etag=?, last_modified=?, last_polled_at=?, last_status=?
+ WHERE id=?""",
+ (state.get("etag"), state.get("last_modified"),
+ datetime.now(UTC).isoformat(), state.get("status", "ok"), src["id"]),
+ )
+
+ # Embed and cluster anything not yet in a cluster, oldest first.
+ pending = con.execute(
+ """SELECT i.* FROM items i
+ LEFT JOIN cluster_members cm ON cm.item_id = i.id
+ WHERE cm.item_id IS NULL
+ ORDER BY i.published_at"""
+ ).fetchall()
+ if pending:
+ pairs = [(r["content_hash"], embed_text(r["title"], r["summary"])) for r in pending]
+ before = con.execute("SELECT COUNT(*) AS n FROM embeddings").fetchone()["n"]
+ vectors = embedder.encode_cached(con, pairs)
+ report.embedded = con.execute(
+ "SELECT COUNT(*) AS n FROM embeddings"
+ ).fetchone()["n"] - before
+
+ assignments = cluster.assign(con, list(pending), vectors)
+ report.clustered = len(assignments)
+ report.new_clusters = sum(1 for a in assignments if a.created)
+ return report
diff --git a/news/sources/__init__.py b/news/sources/__init__.py
new file mode 100644
index 0000000..ae80ddc
--- /dev/null
+++ b/news/sources/__init__.py
@@ -0,0 +1,21 @@
+"""Source adapters. Each returns normalised item dicts; the store does the rest."""
+
+from __future__ import annotations
+
+from . import hn, rss
+
+# No Reddit adapter: the .json endpoints 403 without OAuth as of 2026-09, and
+# the .rss views parse fine as ordinary Atom. Subreddits are configured as
+# `kind = "rss"` instead.
+ADAPTERS = {"rss": rss.fetch, "hn": hn.fetch}
+
+
+async def fetch(client, source) -> tuple[list[dict], dict]:
+ """Dispatch to the adapter for `source['kind']`.
+
+ Returns (items, state) where state carries etag/last_modified to persist.
+ """
+ adapter = ADAPTERS.get(source["kind"])
+ if adapter is None:
+ raise ValueError(f"unknown source kind: {source['kind']!r}")
+ return await adapter(client, source)
diff --git a/news/sources/hn.py b/news/sources/hn.py
new file mode 100644
index 0000000..a859744
--- /dev/null
+++ b/news/sources/hn.py
@@ -0,0 +1,52 @@
+"""Hacker News via the Algolia search API.
+
+Uses `search_by_date` rather than the front page so items are seen early, and
+carries points and comment counts as `ext_*`. Those are re-polled over an item's
+life: an HN post at 3 points is noise and the same post at 300 is news, so the
+routing layer waits for a maturation window instead of firing on first sight.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from ..embed import embed_text
+from ..util import canonical_url, content_hash, stable_id
+
+API = "https://hn.algolia.com/api/v1/search_by_date"
+
+
+async def fetch(client, source) -> tuple[list[dict], dict]:
+ params = {
+ "tags": source.get("query", "story"),
+ "hitsPerPage": str(source.get("page_size", 100)),
+ }
+ resp = await client.get(API, params=params)
+ resp.raise_for_status()
+ now = datetime.now(UTC).isoformat()
+
+ items = []
+ for hit in resp.json().get("hits", []):
+ title = " ".join((hit.get("title") or "").split())
+ # Ask HN and text posts have no URL; point at the discussion instead.
+ link = hit.get("url") or f"https://news.ycombinator.com/item?id={hit['objectID']}"
+ if not title:
+ continue
+ canon = canonical_url(link)
+ items.append(
+ {
+ "id": stable_id(source["id"], canon),
+ "source_id": source["id"],
+ "title": title,
+ "summary": (hit.get("story_text") or "")[:600],
+ "url": link,
+ "canonical_url": canon,
+ "published_at": datetime.fromtimestamp(hit["created_at_i"], UTC).isoformat(),
+ "first_seen_at": now,
+ "content_hash": content_hash(embed_text(title, "")),
+ "ext_score": hit.get("points"),
+ "ext_comments": hit.get("num_comments"),
+ "ext_updated_at": now,
+ }
+ )
+ return items, {"etag": None, "last_modified": None, "status": str(resp.status_code)}
diff --git a/news/sources/rss.py b/news/sources/rss.py
new file mode 100644
index 0000000..a9be754
--- /dev/null
+++ b/news/sources/rss.py
@@ -0,0 +1,100 @@
+"""RSS/Atom via feedparser.
+
+feedparser is used instead of hand-rolled XML parsing on purpose. The old
+system's 700 lines of quick-xml could not read RSS 1.0/RDF at all, which is why
+Deutsche Welle was excluded with the note that it "would yield zero items
+silently". feedparser reads RDF, RSS and Atom, and arXiv and GitHub release
+feeds are just Atom, so they need no adapter of their own.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from email.utils import parsedate_to_datetime
+
+import feedparser
+
+from ..embed import embed_text
+from ..util import canonical_url, content_hash, stable_id
+
+MAX_BYTES = 10 * 1024 * 1024
+
+
+def _published(entry) -> datetime:
+ for key in ("published", "updated", "created"):
+ raw = entry.get(key)
+ if not raw:
+ continue
+ try:
+ dt = parsedate_to_datetime(raw)
+ except (TypeError, ValueError):
+ try:
+ dt = datetime.fromisoformat(raw.replace("Z", "+00:00"))
+ except (TypeError, ValueError):
+ continue
+ return dt if dt.tzinfo else dt.replace(tzinfo=UTC)
+ return datetime.now(UTC)
+
+
+def _summary(entry) -> str:
+ import re
+
+ raw = entry.get("summary") or ""
+ if not raw and entry.get("content"):
+ raw = entry["content"][0].get("value", "")
+ return " ".join(re.sub(r"<[^>]+>", " ", raw).split())
+
+
+async def fetch(client, source) -> tuple[list[dict], dict]:
+ headers = {}
+ if source.get("etag"):
+ headers["If-None-Match"] = source["etag"]
+ if source.get("last_modified"):
+ headers["If-Modified-Since"] = source["last_modified"]
+
+ resp = await client.get(source["url"], headers=headers)
+ state = {
+ "etag": resp.headers.get("etag"),
+ "last_modified": resp.headers.get("last-modified"),
+ "status": str(resp.status_code),
+ }
+ if resp.status_code == 304:
+ return [], state
+
+ # Rate limiting is an expected steady state, not a failure. Reddit in
+ # particular 429s readily on its .rss views. Surface it as a state so the
+ # caller can back off, rather than raising and logging an error every cycle.
+ if resp.status_code == 429:
+ state["retry_after"] = resp.headers.get("retry-after")
+ state["rate_limited"] = True
+ return [], state
+
+ resp.raise_for_status()
+
+ parsed = feedparser.parse(resp.content[:MAX_BYTES])
+ now = datetime.now(UTC).isoformat()
+ items = []
+ for entry in parsed.entries:
+ link = entry.get("link") or ""
+ title = " ".join((entry.get("title") or "").split())
+ if not link or not title:
+ continue
+ canon = canonical_url(link)
+ summary = _summary(entry)
+ items.append(
+ {
+ "id": stable_id(source["id"], canon),
+ "source_id": source["id"],
+ "title": title,
+ "summary": summary,
+ "url": link,
+ "canonical_url": canon,
+ "published_at": _published(entry).isoformat(),
+ "first_seen_at": now,
+ "content_hash": content_hash(embed_text(title, summary)),
+ "ext_score": None,
+ "ext_comments": None,
+ "ext_updated_at": None,
+ }
+ )
+ return items, state
diff --git a/news/store.py b/news/store.py
new file mode 100644
index 0000000..f723d14
--- /dev/null
+++ b/news/store.py
@@ -0,0 +1,170 @@
+"""SQLite schema and repository helpers.
+
+Text ids and RFC3339 timestamps throughout, so the database stays readable with
+the `sqlite3` CLI. Vectors are stored as raw float32 blobs.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+from collections.abc import Iterable, Sequence
+from pathlib import Path
+
+import numpy as np
+
+SCHEMA = """
+CREATE TABLE IF NOT EXISTS sources (
+ id TEXT PRIMARY KEY,
+ name TEXT NOT NULL,
+ url TEXT NOT NULL UNIQUE,
+ kind TEXT NOT NULL, -- rss | hn | reddit
+ -- Editorial-independence bucket. Importance counts DISTINCT buckets, not
+ -- feeds: ten outlets reprinting one wire story is one bucket, not ten.
+ bucket TEXT NOT NULL,
+ weight REAL NOT NULL DEFAULT 1.0,
+ -- 1 for a publisher's front-page/top-stories feed. Appearing there is a free
+ -- editorial-prominence signal; the old system dropped it by unsubscribing.
+ prominence INTEGER NOT NULL DEFAULT 0,
+ enabled INTEGER NOT NULL DEFAULT 1,
+ poll_interval_secs INTEGER NOT NULL DEFAULT 900,
+ etag TEXT,
+ last_modified TEXT,
+ last_polled_at TEXT,
+ last_status TEXT
+);
+
+CREATE TABLE IF NOT EXISTS items (
+ id TEXT PRIMARY KEY,
+ source_id TEXT NOT NULL REFERENCES sources(id),
+ title TEXT NOT NULL,
+ summary TEXT NOT NULL DEFAULT '',
+ url TEXT NOT NULL,
+ canonical_url TEXT NOT NULL,
+ published_at TEXT NOT NULL,
+ first_seen_at TEXT NOT NULL,
+ content_hash TEXT NOT NULL,
+ -- Aggregator-native signal (HN points, Reddit score). Re-polled over the
+ -- item's life: an HN post at 3 points is noise, at 300 it is news.
+ ext_score INTEGER,
+ ext_comments INTEGER,
+ ext_updated_at TEXT,
+ UNIQUE(source_id, canonical_url)
+);
+CREATE INDEX IF NOT EXISTS items_published ON items(published_at);
+CREATE INDEX IF NOT EXISTS items_hash ON items(content_hash);
+
+-- Keyed by content hash, not item id, so identical text across feeds is
+-- embedded exactly once. Embedding is the slow step (~11 items/sec).
+CREATE TABLE IF NOT EXISTS embeddings (
+ content_hash TEXT PRIMARY KEY,
+ dim INTEGER NOT NULL,
+ vec BLOB NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS clusters (
+ id TEXT PRIMARY KEY,
+ centroid BLOB NOT NULL,
+ size INTEGER NOT NULL DEFAULT 0,
+ first_seen_at TEXT NOT NULL,
+ last_seen_at TEXT NOT NULL,
+ lead_item_id TEXT REFERENCES items(id),
+ headline TEXT
+);
+CREATE INDEX IF NOT EXISTS clusters_last_seen ON clusters(last_seen_at);
+
+CREATE TABLE IF NOT EXISTS cluster_members (
+ cluster_id TEXT NOT NULL REFERENCES clusters(id) ON DELETE CASCADE,
+ item_id TEXT NOT NULL REFERENCES items(id),
+ added_at TEXT NOT NULL,
+ PRIMARY KEY (cluster_id, item_id)
+);
+CREATE INDEX IF NOT EXISTS cluster_members_item ON cluster_members(item_id);
+
+-- Append-only: every click is a training event, never an update in place.
+CREATE TABLE IF NOT EXISTS feedback (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ cluster_id TEXT NOT NULL,
+ kind TEXT NOT NULL, -- interested | not_interested
+ source TEXT NOT NULL DEFAULT 'web', -- web | ntfy
+ created_at TEXT NOT NULL
+);
+"""
+
+
+def connect(path: str | Path) -> sqlite3.Connection:
+ path = Path(path)
+ path.parent.mkdir(parents=True, exist_ok=True)
+ con = sqlite3.connect(path, isolation_level=None)
+ con.row_factory = sqlite3.Row
+ con.execute("PRAGMA journal_mode=WAL")
+ con.execute("PRAGMA foreign_keys=ON")
+ con.executescript(SCHEMA)
+ return con
+
+
+def to_blob(vec: np.ndarray) -> bytes:
+ return np.asarray(vec, dtype=np.float32).tobytes()
+
+
+def from_blob(blob: bytes) -> np.ndarray:
+ return np.frombuffer(blob, dtype=np.float32)
+
+
+def upsert_source(con: sqlite3.Connection, **kw) -> None:
+ """Config is authoritative for everything except the polling state columns."""
+ con.execute(
+ """INSERT INTO sources (id,name,url,kind,bucket,weight,prominence,
+ enabled,poll_interval_secs)
+ VALUES (:id,:name,:url,:kind,:bucket,:weight,:prominence,
+ :enabled,:poll_interval_secs)
+ ON CONFLICT(url) DO UPDATE SET
+ name=excluded.name, kind=excluded.kind, bucket=excluded.bucket,
+ weight=excluded.weight, prominence=excluded.prominence,
+ enabled=excluded.enabled,
+ poll_interval_secs=excluded.poll_interval_secs""",
+ kw,
+ )
+
+
+def insert_items(con: sqlite3.Connection, rows: Iterable[dict]) -> int:
+ """Returns the number of genuinely NEW items.
+
+ Counted by row count, not `total_changes`: the ON CONFLICT branch updates
+ aggregator scores on every re-poll, and counting those as new would make an
+ HN source look like it produced hundreds of fresh items an hour.
+ """
+ before = con.execute("SELECT COUNT(*) AS n FROM items").fetchone()["n"]
+ con.executemany(
+ """INSERT INTO items (id,source_id,title,summary,url,canonical_url,
+ published_at,first_seen_at,content_hash,
+ ext_score,ext_comments,ext_updated_at)
+ VALUES (:id,:source_id,:title,:summary,:url,:canonical_url,
+ :published_at,:first_seen_at,:content_hash,
+ :ext_score,:ext_comments,:ext_updated_at)
+ ON CONFLICT(source_id,canonical_url) DO UPDATE SET
+ ext_score=excluded.ext_score,
+ ext_comments=excluded.ext_comments,
+ ext_updated_at=excluded.ext_updated_at
+ WHERE excluded.ext_score IS NOT NULL""",
+ list(rows),
+ )
+ return con.execute("SELECT COUNT(*) AS n FROM items").fetchone()["n"] - before
+
+
+def cached_vectors(con: sqlite3.Connection, hashes: Sequence[str]) -> dict[str, np.ndarray]:
+ out: dict[str, np.ndarray] = {}
+ for i in range(0, len(hashes), 500):
+ chunk = hashes[i : i + 500]
+ q = ",".join("?" * len(chunk))
+ for row in con.execute(
+ f"SELECT content_hash, vec FROM embeddings WHERE content_hash IN ({q})", chunk
+ ):
+ out[row["content_hash"]] = from_blob(row["vec"])
+ return out
+
+
+def store_vectors(con: sqlite3.Connection, pairs: Iterable[tuple[str, np.ndarray]]) -> None:
+ con.executemany(
+ "INSERT OR REPLACE INTO embeddings (content_hash,dim,vec) VALUES (?,?,?)",
+ [(h, len(v), to_blob(v)) for h, v in pairs],
+ )
diff --git a/news/util.py b/news/util.py
new file mode 100644
index 0000000..68b0962
--- /dev/null
+++ b/news/util.py
@@ -0,0 +1,55 @@
+"""URL canonicalisation and stable hashing.
+
+The cheap deterministic wins that need no model: the same article arrives from
+several feeds with different tracking parameters.
+
+Note what is deliberately absent. A SimHash near-duplicate stage was written and
+then removed after measurement: on real news it gave +1.2 bits of separation
+between same-story and random pairs on titles, and -0.2 with summaries included.
+News duplicates are *paraphrases* ("Explosion at Bolivia military base kills at
+least 10" / "Blasts at Bolivian military base kill at least 10, injure dozens"),
+which share almost no word trigrams. That is the same lexical trap that made the
+old system's Jaccard-on-title-tokens clustering fail. Embeddings separate the
+identical group at +0.353 cosine, so dedup lives in `cluster.py` instead.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import re
+import uuid
+from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
+
+# uuid5 namespace for this project, so ids are stable across rebuilds.
+NS = uuid.UUID("6f3b1e2a-9c4d-5f8a-b7e1-2d4c6a8b0f31")
+
+# Tracking parameters carried by feed links. Stripping them is what makes the
+# same article from two feeds compare equal.
+_JUNK_PARAMS = re.compile(
+ r"^(utm_|ito$|ns_|cmp$|CMP$|fbclid$|gclid$|igshid$|mc_[ce]id$|ref$|"
+ r"ref_src$|s_kwcid$|smid$|partner$|at_(medium|campaign|custom\d)$|"
+ r"__twitter_impression$|guccounter$|sh$|amp$)"
+)
+
+
+def canonical_url(url: str) -> str:
+ """Strip tracking noise so the same article from two feeds compares equal."""
+ parts = urlsplit(url.strip())
+ host = parts.netloc.lower().removeprefix("www.")
+ query = urlencode(
+ [(k, v) for k, v in parse_qsl(parts.query, keep_blank_values=True)
+ if not _JUNK_PARAMS.match(k)]
+ )
+ path = parts.path.rstrip("/") or "/"
+ # Fragments never identify a distinct article in a feed context.
+ return urlunsplit((parts.scheme.lower() or "https", host, path, query, ""))
+
+
+def stable_id(*parts: str) -> str:
+ """Deterministic uuid5 so re-ingesting the same item is idempotent."""
+ return str(uuid.uuid5(NS, "\x1f".join(parts)))
+
+
+def content_hash(text: str) -> str:
+ """Key for the embedding cache: identical text is embedded exactly once."""
+ return hashlib.blake2b(text.encode("utf-8"), digest_size=16).hexdigest()
diff --git a/profile/profile.toml b/profile/profile.toml
index 1424d0f..b25b2a6 100644
--- a/profile/profile.toml
+++ b/profile/profile.toml
@@ -203,6 +203,13 @@ probes = [
# 3,885 items in the v0.2 database and Connor's 40 real feedback labels.
# See profile/CALIBRATION.md for the measurements.
+[delivery]
+# Settled with Connor 2026-09-08.
+digest_hour = "08:45"
+quiet_hours = { start = "22:00", end = "07:00" }
+# The importance lane MAY override quiet hours. The interest lane never may.
+importance_may_wake = true
+
[scoring]
# Compare blocks by z-score WITHIN each block, never by raw cosine. Raw cosine
# is not comparable across blocks -- "geopolitics" averages 0.501 similarity to
diff --git a/profile/sources.toml b/profile/sources.toml
new file mode 100644
index 0000000..5d270e7
--- /dev/null
+++ b/profile/sources.toml
@@ -0,0 +1,268 @@
+# Sources.
+#
+# `bucket` is the editorial-independence group, and it is the important field.
+# Importance counts DISTINCT buckets, never feeds: two BBC feeds are one bucket,
+# and ten outlets reprinting one wire story must not read as ten independent
+# confirmations. Getting this wrong is what made the old system fire priority-5
+# pushes on anything two feeds happened to carry.
+#
+# `prominence = 1` marks a publisher's front-page/top-stories feed. Subscribing
+# to both it and the section feed is a free editorial-promotion signal; the old
+# config dropped BBC's top-news feed and threw that away.
+
+[defaults]
+poll_interval_secs = 900
+
+# ------------------------------------------------------------- world / wire --
+[[source]]
+name = "Al Jazeera"
+url = "https://www.aljazeera.com/xml/rss/all.xml"
+bucket = "aljazeera"
+weight = 1.2
+
+[[source]]
+name = "BBC Top Stories"
+url = "https://feeds.bbci.co.uk/news/rss.xml"
+bucket = "bbc"
+prominence = 1
+weight = 1.1
+
+[[source]]
+name = "BBC World"
+url = "https://feeds.bbci.co.uk/news/world/rss.xml"
+bucket = "bbc"
+weight = 1.1
+
+[[source]]
+name = "Guardian World"
+url = "https://www.theguardian.com/world/rss"
+bucket = "guardian"
+
+[[source]]
+name = "NYT World"
+url = "https://rss.nytimes.com/services/xml/rss/nyt/World.xml"
+bucket = "nyt"
+
+[[source]]
+name = "NPR News"
+url = "https://feeds.npr.org/1001/rss.xml"
+bucket = "npr"
+
+[[source]]
+name = "France24"
+url = "https://www.france24.com/en/rss"
+bucket = "france24"
+
+# RSS 1.0/RDF. The old hand-rolled quick-xml parser could not read this format
+# and DW was excluded with the note that it "would yield zero items silently".
+# feedparser handles RDF, so DW is back.
+[[source]]
+name = "Deutsche Welle"
+url = "https://rss.dw.com/rdf/rss-en-all"
+bucket = "dw"
+
+[[source]]
+name = "ABC Australia"
+url = "https://www.abc.net.au/news/feed/51120/rss.xml"
+bucket = "abc-au"
+
+[[source]]
+name = "PBS NewsHour"
+url = "https://www.pbs.org/newshour/feeds/rss/headlines"
+bucket = "pbs"
+poll_interval_secs = 1800
+
+# --------------------------------------------------------------- mideast ----
+[[source]]
+name = "Middle East Eye"
+url = "https://www.middleeasteye.net/rss"
+bucket = "middle-east-eye"
+
+[[source]]
+name = "+972 Magazine"
+url = "https://www.972mag.com/feed/"
+bucket = "972mag"
+poll_interval_secs = 3600
+
+# ----------------------------------------------------------- tech / policy --
+[[source]]
+name = "Ars Technica"
+url = "https://feeds.arstechnica.com/arstechnica/index"
+bucket = "ars-technica"
+
+[[source]]
+name = "EFF Deeplinks"
+url = "https://www.eff.org/rss/updates.xml"
+bucket = "eff"
+poll_interval_secs = 3600
+
+# Global-South technology and economics; covers two interest blocks at once.
+[[source]]
+name = "Rest of World"
+url = "https://restofworld.org/feed/latest/"
+bucket = "rest-of-world"
+poll_interval_secs = 3600
+
+# ------------------------------------------------------ software / systems --
+[[source]]
+name = "Hacker News"
+url = "https://hn.algolia.com/api/v1/search_by_date"
+kind = "hn"
+bucket = "hacker-news"
+poll_interval_secs = 1800
+
+[[source]]
+name = "Lobsters"
+url = "https://lobste.rs/rss"
+bucket = "lobsters"
+poll_interval_secs = 1800
+
+# Reddit's .json endpoints return 403 without OAuth as of 2026-09; the .rss
+# views work unauthenticated but 429 aggressively -- three requests in quick
+# succession from one IP were enough to get throttled during validation, and the
+# block persisted across serialized retries. Hence the long intervals here.
+# The fetcher treats 429 as a state to back off from, not an error.
+#
+# Cost of the .rss route: the Atom feed carries no score, so these contribute no
+# aggregator-maturity signal. Hacker News is the source that supplies it.
+[[source]]
+name = "r/LocalLLaMA"
+url = "https://www.reddit.com/r/LocalLLaMA/new/.rss"
+bucket = "reddit"
+poll_interval_secs = 5400
+
+[[source]]
+name = "r/selfhosted"
+url = "https://www.reddit.com/r/selfhosted/new/.rss"
+bucket = "reddit"
+poll_interval_secs = 7200
+
+[[source]]
+name = "r/rust"
+url = "https://www.reddit.com/r/rust/new/.rss"
+bucket = "reddit"
+poll_interval_secs = 7200
+
+# Release feeds for the stack Connor actually runs. GitHub releases are Atom,
+# so they need no adapter of their own.
+[[source]]
+name = "llama.cpp releases"
+url = "https://github.com/ggml-org/llama.cpp/releases.atom"
+bucket = "github-releases"
+poll_interval_secs = 7200
+
+[[source]]
+name = "LiteLLM releases"
+url = "https://github.com/BerriAI/litellm/releases.atom"
+bucket = "github-releases"
+poll_interval_secs = 7200
+
+[[source]]
+name = "llama-swap releases"
+url = "https://github.com/mostlygeek/llama-swap/releases.atom"
+bucket = "github-releases"
+poll_interval_secs = 7200
+
+[[source]]
+name = "Podman releases"
+url = "https://github.com/containers/podman/releases.atom"
+bucket = "github-releases"
+poll_interval_secs = 7200
+
+[[source]]
+name = "Rust releases"
+url = "https://github.com/rust-lang/rust/releases.atom"
+bucket = "github-releases"
+poll_interval_secs = 7200
+
+# arXiv announces once per weekday (~00:00 UTC); the feed is legitimately EMPTY
+# between announcement windows, which is not a fault. Verified 2026-09-08 that
+# the endpoint serves valid RSS -- cs.CL, cs.LG and cs.AI were all empty at the
+# time of checking, consistent with being between windows.
+[[source]]
+name = "arXiv cs.CL"
+url = "https://rss.arxiv.org/rss/cs.CL"
+bucket = "arxiv"
+poll_interval_secs = 21600
+
+# ------------------------------------------------------------------ labor ---
+[[source]]
+name = "Labor Notes"
+url = "https://labornotes.org/feed"
+bucket = "labor-notes"
+poll_interval_secs = 3600
+
+[[source]]
+name = "Jacobin"
+url = "https://jacobin.com/feed/"
+bucket = "jacobin"
+poll_interval_secs = 3600
+
+# --------------------------------------------------------------- urbanism ---
+[[source]]
+name = "Streetsblog USA"
+url = "https://usa.streetsblog.org/feed"
+bucket = "streetsblog"
+poll_interval_secs = 3600
+
+[[source]]
+name = "Strong Towns"
+url = "https://www.strongtowns.org/journal?format=rss"
+bucket = "strong-towns"
+poll_interval_secs = 7200
+
+# --------------------------------------------------------- energy / climate --
+[[source]]
+name = "Canary Media"
+url = "https://www.canarymedia.com/rss"
+bucket = "canary-media"
+poll_interval_secs = 3600
+
+# --------------------------------------------------------------- US / local --
+[[source]]
+name = "ProPublica"
+url = "https://www.propublica.org/feeds/propublica/main"
+bucket = "propublica"
+poll_interval_secs = 3600
+
+[[source]]
+name = "NPR Politics"
+url = "https://feeds.npr.org/1014/rss.xml"
+bucket = "npr"
+
+[[source]]
+name = "Ohio Capital Journal"
+url = "https://ohiocapitaljournal.com/feed/"
+bucket = "ohio-capital-journal"
+poll_interval_secs = 3600
+
+[[source]]
+name = "WVXU Cincinnati"
+url = "https://www.wvxu.org/local-news.rss"
+bucket = "wvxu"
+poll_interval_secs = 3600
+
+[[source]]
+name = "WCPO Cincinnati"
+url = "https://www.wcpo.com/news.rss"
+bucket = "wcpo"
+poll_interval_secs = 3600
+
+[[source]]
+name = "Kentucky Lantern"
+url = "https://kentuckylantern.com/feed/"
+bucket = "kentucky-lantern"
+poll_interval_secs = 3600
+
+# ------------------------------------------------------------ italy / mexico --
+[[source]]
+name = "ANSA English"
+url = "https://www.ansa.it/english/english_rss.xml"
+bucket = "ansa"
+poll_interval_secs = 3600
+
+[[source]]
+name = "Mexico News Daily"
+url = "https://mexiconewsdaily.com/feed/"
+bucket = "mexico-news-daily"
+poll_interval_secs = 3600
diff --git a/pyproject.toml b/pyproject.toml
index 4b47540..16d60e2 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -26,3 +26,4 @@ build-backend = "hatchling.build"
[tool.pytest.ini_options]
testpaths = ["tests"]
+asyncio_mode = "auto"
diff --git a/tests/test_cluster.py b/tests/test_cluster.py
new file mode 100644
index 0000000..5d369ff
--- /dev/null
+++ b/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"
diff --git a/tests/test_rss.py b/tests/test_rss.py
new file mode 100644
index 0000000..7e0f5c8
--- /dev/null
+++ b/tests/test_rss.py
@@ -0,0 +1,81 @@
+"""RSS/Atom parsing, including the RDF format the old system could not read."""
+
+import httpx
+import pytest
+
+from news.sources import rss
+
+RDF = """
+
+ Thttp://x
+ d
+ -
+ German economy frets about election result
+ http://x/1?utm_source=rss
+ Some <b>markup</b> here
+
+"""
+
+ATOM = """
+
+ F
+
+ Release v1.2.3
+
+
+ 2026-09-08T12:00:00Z
+
+"""
+
+
+def _source():
+ return {"id": "s1", "url": "http://feed", "kind": "rss"}
+
+
+async def _fetch(body: str, status: int = 200, headers: dict | None = None):
+ def handler(request):
+ return httpx.Response(status, content=body.encode(), headers=headers or {})
+
+ transport = httpx.MockTransport(handler)
+ async with httpx.AsyncClient(transport=transport) as client:
+ return await rss.fetch(client, _source())
+
+
+@pytest.mark.asyncio
+async def test_parses_rss_1_rdf():
+ """DW ships RDF; the old hand-rolled parser yielded zero items silently."""
+ items, _ = await _fetch(RDF)
+ assert len(items) == 1
+ assert items[0]["title"].startswith("German economy")
+ # tracking parameter stripped, markup removed from the summary
+ assert items[0]["canonical_url"] == "http://x/1"
+ assert "" not in items[0]["summary"] and "markup" in items[0]["summary"]
+
+
+@pytest.mark.asyncio
+async def test_atom_prefers_alternate_link_over_self():
+ items, _ = await _fetch(ATOM)
+ assert len(items) == 1
+ assert items[0]["canonical_url"] == "http://x/rel"
+
+
+@pytest.mark.asyncio
+async def test_304_returns_no_items_and_is_not_an_error():
+ items, state = await _fetch("", status=304)
+ assert items == [] and state["status"] == "304"
+
+
+@pytest.mark.asyncio
+async def test_429_is_reported_as_rate_limited_not_raised():
+ """Reddit 429s readily; that must back off, not raise every cycle."""
+ items, state = await _fetch("", status=429, headers={"retry-after": "120"})
+ assert items == []
+ assert state["rate_limited"] is True
+ assert state["retry_after"] == "120"
+
+
+@pytest.mark.asyncio
+async def test_server_error_still_raises():
+ with pytest.raises(httpx.HTTPStatusError):
+ await _fetch("", status=503)
diff --git a/tests/test_util.py b/tests/test_util.py
new file mode 100644
index 0000000..8c37a60
--- /dev/null
+++ b/tests/test_util.py
@@ -0,0 +1,26 @@
+from news.util import canonical_url, content_hash, stable_id
+
+
+def test_canonical_url_strips_tracking_and_www():
+ a = canonical_url("https://www.bbc.co.uk/news/world-123?utm_source=rss&utm_medium=feed")
+ b = canonical_url("https://bbc.co.uk/news/world-123/")
+ assert a == b == "https://bbc.co.uk/news/world-123"
+
+
+def test_canonical_url_keeps_meaningful_query():
+ url = canonical_url("https://news.ycombinator.com/item?id=4242&utm_campaign=x")
+ assert url == "https://news.ycombinator.com/item?id=4242"
+
+
+def test_canonical_url_drops_fragment():
+ assert canonical_url("https://x.com/a#section") == "https://x.com/a"
+
+
+def test_stable_id_is_deterministic_and_distinct():
+ assert stable_id("source", "u") == stable_id("source", "u")
+ assert stable_id("source", "u") != stable_id("source", "v")
+
+
+def test_content_hash_matches_only_identical_text():
+ assert content_hash("a b") == content_hash("a b")
+ assert content_hash("a b") != content_hash("a b")