diff --git a/README.md b/README.md index a28d6e4..df9f523 100644 --- a/README.md +++ b/README.md @@ -73,6 +73,19 @@ corpus with a specific embedding model. **Changing `[embedding].model` invalidat both**, because a different model puts cosine similarity on a different scale. Re-run the calibration in `profile/CALIBRATION.md` before trusting them. +## Commands + + news poll one fetch -> embed -> cluster cycle + news sources sources and item counts + news clusters inspect clusters and their corroboration + news score score clusters on both axes + news simulate what would these thresholds actually have sent? + +`simulate` is the tuning surface. The old system's replay harness hardcoded its +parameters and so could never sweep the thresholds it existed to tune. + ## Status -Early. `profile/` is written; the pipeline is not yet built. +Ingest, embedding, clustering, two-axis scoring and routing are built and +running against live feeds. Delivery (ntfy, the digest page) and the web UI are +not yet written. diff --git a/news/cli.py b/news/cli.py index 9b9b315..9a50c24 100644 --- a/news/cli.py +++ b/news/cli.py @@ -68,6 +68,81 @@ def cmd_clusters(args) -> int: return 0 +def _score_all(con, cfg): + from .score import ProfileVectors, score_clusters + pv = ProfileVectors(con, cfg.profile(), Embedder.from_config(cfg)) + return pv, score_clusters(con, pv, cfg.scoring, cfg.clustering) + + +def cmd_score(args) -> int: + cfg = Config() + con = store.connect(cfg.db_path) + _, scores = _score_all(con, cfg) + scores.sort(key=lambda s: -max(s.interest_score, s.importance_score)) + print(f"{'inter':>6} {'impor':>6} {'block':<24} {'flags':<14} story") + print("-" * 108) + for s in scores[: args.limit]: + flags = [] + if s.veto: + flags.append("VETO") + if s.demote: + flags.append(f"-{s.demote_penalty:g}") + print(f"{s.interest_score:>6.2f} {s.importance_score:>6.2f} " + f"{s.interest.block[:23]:<24} {','.join(flags)[:13]:<14} {s.lead_title[:52]}") + return 0 + + +def cmd_simulate(args) -> int: + """What would these settings actually have sent? + + The tuning surface the old system never had: its replay harness hardcoded + default parameters and so could never sweep the thresholds it existed to + tune. + """ + from datetime import UTC, datetime + + from .route import DIGEST, DROP, IMPORTANT, INTEREST, route + + cfg = Config() + con = store.connect(cfg.db_path) + _, scores = _score_all(con, cfg) + routing = dict(cfg.routing) + if args.interest_threshold is not None: + routing["interest_threshold"] = args.interest_threshold + if args.important_threshold is not None: + routing["important_threshold"] = args.important_threshold + + now = datetime.now(UTC) + decisions = route(con, scores, routing, cfg.delivery, now) + by_lane = {} + for d in decisions: + by_lane.setdefault(d.lane, []).append(d) + + # Rate is over the notification-eligible window, not the whole archive: + # backfilled items inflate the span and would understate the true rate. + days = routing["max_age_hours"] / 24.0 + eligible = sum(1 for s in scores + if s.last_seen and (now - s.last_seen).total_seconds() + <= routing["max_age_hours"] * 3600) + print(f"{len(scores)} clusters, {eligible} inside the " + f"{routing['max_age_hours']}h notification window ({days:.1f} days)") + print(f"thresholds: interest {routing['interest_threshold']}, " + f"important {routing['important_threshold']}\n") + for lane in (IMPORTANT, INTEREST, DIGEST, DROP): + n = len(by_lane.get(lane, [])) + rate = f"{n / days:5.1f}/day" if lane != DROP else " " * 9 + print(f" {lane:<18} {n:>5} {rate}") + + for lane in (IMPORTANT, INTEREST): + items = sorted(by_lane.get(lane, []), key=lambda d: -d.score) + if items: + print(f"\n=== {lane} ===") + for d in items[: args.show]: + title = next(s.lead_title for s in scores if s.cluster_id == d.cluster_id) + print(f" {d.score:>6.2f} [{d.reason[:28]:<28}] {title[:56]}") + return 0 + + def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(prog="news") sub = parser.add_subparsers(dest="cmd", required=True) @@ -81,6 +156,16 @@ def main(argv: list[str] | None = None) -> int: p.add_argument("-v", "--verbose", action="store_true") p.set_defaults(fn=cmd_clusters) + p = sub.add_parser("score", help="score clusters on both axes") + p.add_argument("--limit", type=int, default=25) + p.set_defaults(fn=cmd_score) + + p = sub.add_parser("simulate", help="what would these thresholds have sent?") + p.add_argument("--interest-threshold", type=float, default=None) + p.add_argument("--important-threshold", type=float, default=None) + p.add_argument("--show", type=int, default=8) + p.set_defaults(fn=cmd_simulate) + args = parser.parse_args(argv) return args.fn(args) diff --git a/news/config.py b/news/config.py index f9e2bbe..b70ab7f 100644 --- a/news/config.py +++ b/news/config.py @@ -97,6 +97,24 @@ class Config: "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), + "demote_absolute": s.get("demote_absolute", 0.55), + "importance_weights": s.get("importance_weights", {}), + } + + @property + def routing(self) -> dict[str, Any]: + r = self.section("routing") + return { + "interest_per_day": r.get("interest_per_day", 3), + "important_per_day": r.get("important_per_day", 2), + "important_per_night": r.get("important_per_night", 1), + "interest_threshold": r.get("interest_threshold", 2.0), + "important_threshold": r.get("important_threshold", 2.0), + "digest_floor": r.get("digest_floor", 0.5), + "hn_min_score": r.get("hn_min_score", 150), + "max_age_hours": r.get("max_age_hours", 48), + "digest_max_items": r.get("digest_max_items", 25), + "serendipity_max_interest": r.get("serendipity_max_interest", 0.5), } # -- reader ------------------------------------------------------------ diff --git a/news/route.py b/news/route.py new file mode 100644 index 0000000..4002fe9 --- /dev/null +++ b/news/route.py @@ -0,0 +1,220 @@ +"""Routing: turn scores into decisions. + +Two independent lanes, because "something I might like" and "something I need to +know" are different questions: + + alert:interest high interest. Respects quiet hours, always. + alert:important high importance regardless of interest. May wake the + reader, with its own smaller budget and a hard nightly cap. + digest everything else worth reading, collected for the morning. + drop vetoed, or below the floor. + +Vetoes are absolute on BOTH lanes. In the old system the bypass lane never +consulted relevance (gate.rs:302), so a blocklisted item could still fire at +priority 5. That must not recur, and a test asserts it. +""" + +from __future__ import annotations + +import sqlite3 +from dataclasses import dataclass +from datetime import datetime, time, timedelta + +from .score import ClusterScore + +INTEREST = "alert:interest" +IMPORTANT = "alert:important" +DIGEST = "digest" +DROP = "drop" + + +@dataclass(frozen=True) +class Decision: + cluster_id: str + lane: str + reason: str + score: float + + +def _parse_hhmm(value: str) -> time: + hh, mm = value.split(":") + return time(int(hh), int(mm)) + + +def in_quiet_hours(now: datetime, quiet: dict) -> bool: + start = _parse_hhmm(quiet.get("start", "22:00")) + end = _parse_hhmm(quiet.get("end", "07:00")) + minute = now.hour * 60 + now.minute + s, e = start.hour * 60 + start.minute, end.hour * 60 + end.minute + # Quiet hours normally wrap midnight. + return minute >= s or minute < e if s > e else s <= minute < e + + +def _sent_since(con: sqlite3.Connection, lane: str, since: datetime) -> int: + row = con.execute( + "SELECT COUNT(*) AS n FROM notification_log WHERE lane=? AND created_at>=?", + (lane, since.isoformat()), + ).fetchone() + return int(row["n"] or 0) + + +def route( + con: sqlite3.Connection, + scores: list[ClusterScore], + routing: dict, + delivery: dict, + now: datetime, +) -> list[Decision]: + """Decide a lane for every scored cluster. + + Clusters are considered strongest-first within each lane, so a limited daily + budget is spent on the best candidates rather than on whatever arrived + first -- the ordering bug that let the old system's budget go to noise. + """ + quiet = delivery.get("quiet_hours", {}) + quiet_now = in_quiet_hours(now, quiet) + may_wake = bool(delivery.get("importance_may_wake", True)) + day_ago = now - timedelta(days=1) + + interest_budget = routing.get("interest_per_day", 3) - _sent_since(con, INTEREST, day_ago) + important_budget = routing.get("important_per_day", 2) - _sent_since(con, IMPORTANT, day_ago) + night_budget = routing.get("important_per_night", 1) - _sent_since( + con, IMPORTANT, now - timedelta(hours=10)) + + already = { + r["cluster_id"] + for r in con.execute( + "SELECT DISTINCT cluster_id FROM notification_log WHERE created_at>=?", + ((now - timedelta(days=3)).isoformat(),), + ) + } + + decisions: dict[str, Decision] = {} + # Deferred items are digest CANDIDATES, not digest members. Everything that + # misses an alert lane competes for the same capped digest, so a quiet-hours + # deferral cannot smuggle 190 items past the cap the way an earlier version + # did. + deferred: dict[str, str] = {} + + def settle(s: ClusterScore, lane: str, reason: str, value: float) -> None: + decisions[s.cluster_id] = Decision(s.cluster_id, lane, reason, value) + + def defer(s: ClusterScore, reason: str) -> None: + deferred[s.cluster_id] = reason + + max_age = timedelta(hours=routing.get("max_age_hours", 48)) + live: list[ClusterScore] = [] + for s in scores: + # Staleness first. Feeds hand back large archives on first fetch -- one + # run pulled 88 items going back five months, plus a BBC entry literally + # titled "BBC News app" dated 2025. Those belong in the scoring corpus, + # because the z-score population should reflect what news looks like, + # but nothing that old may reach the reader. + if s.last_seen is not None and now - s.last_seen > max_age: + settle(s, DROP, "stale", s.interest_score) + elif s.veto: + settle(s, DROP, f"veto:{s.veto}", s.interest_score) + elif s.cluster_id in already: + settle(s, DROP, "already-notified", s.interest_score) + elif _immature(s, routing): + # Not rejected, just not ripe: the crowd has not voted yet. It is + # re-scored on the next cycle as its score climbs. + settle(s, DROP, "aggregator-immature", s.interest_score) + else: + live.append(s) + + # --- importance lane first: smaller budget, higher stakes --- + for s in sorted(live, key=lambda x: -x.importance_score): + if s.cluster_id in decisions or s.cluster_id in deferred: + continue + if s.importance_score < routing.get("important_threshold", 2.0): + continue + if important_budget <= 0: + defer(s, "important-budget-exhausted") + continue + if quiet_now: + if not may_wake or night_budget <= 0: + defer(s, "quiet-hours") + continue + night_budget -= 1 + important_budget -= 1 + settle(s, IMPORTANT, f"importance:{s.importance.block}", s.importance_score) + + # --- interest lane: never overrides quiet hours --- + for s in sorted(live, key=lambda x: -x.interest_score): + if s.cluster_id in decisions or s.cluster_id in deferred: + continue + if s.interest_score < routing.get("interest_threshold", 2.0): + continue + if _conflict_needs_importance(s, routing, delivery): + defer(s, "conflict-general-needs-importance") + continue + if quiet_now: + defer(s, "quiet-hours") + continue + if interest_budget <= 0: + defer(s, "interest-budget-exhausted") + continue + interest_budget -= 1 + settle(s, INTEREST, f"interest:{s.interest.block}", s.interest_score) + + # --- one capped, ranked digest --- + floor = routing.get("digest_floor", 0.0) + cap = routing.get("digest_max_items", 25) + pool = sorted( + (s for s in live if s.cluster_id not in decisions), + key=lambda x: -max(x.interest_score, x.importance_score), + ) + taken = 0 + for s in pool: + value = max(s.interest_score, s.importance_score) + reason = deferred.get(s.cluster_id, "digest") + if value < floor and s.cluster_id not in deferred: + settle(s, DROP, "below-floor", value) + elif taken < cap: + settle(s, DIGEST, reason, value) + taken += 1 + else: + settle(s, DROP, "digest-full", value) + + # Serendipity: reserve slots for items that scored well on importance but + # POORLY on the interest profile, so the digest does not become a mirror of + # the profile. A preference learner tightens its own bubble as it succeeds. + slots = int(delivery.get("serendipity_slots", 0)) + if slots: + ceiling = routing.get("serendipity_max_interest", 0.5) + outside = sorted( + (s for s in pool + if decisions[s.cluster_id].lane == DROP and s.interest_score < ceiling), + key=lambda x: -x.importance_score, + ) + for s in outside[:slots]: + settle(s, DIGEST, "serendipity", s.importance_score) + + return [decisions[s.cluster_id] for s in scores] + + +def _immature(s: ClusterScore, routing: dict) -> bool: + """Aggregator items wait for the crowd. + + An HN post at 3 points is noise and the same post at 300 is news, so an item + whose only signal is an aggregator score below the bar is held for the + digest rather than routed on arrival. + """ + threshold = routing.get("hn_min_score", 150) + return s.ext_score is not None and s.ext_score < threshold and s.corroboration.origins <= 1 + + +def _conflict_needs_importance(s: ClusterScore, routing: dict, delivery: dict) -> bool: + """Category-E conflict notifies only on a major development. + + Connor: "my actual war interest is low, but unfortunately all wars are + important." Routine front-line reporting goes to the digest; a new + offensive, a ceasefire or mass casualties clears the importance bar and + reaches him through the other lane. + """ + if not delivery.get("conflict_general_requires_importance", True): + return False + if s.interest.block != "conflict-general": + return False + return s.importance_score < routing.get("important_threshold", 2.0) diff --git a/news/score.py b/news/score.py new file mode 100644 index 0000000..a5e9d33 --- /dev/null +++ b/news/score.py @@ -0,0 +1,247 @@ +"""Two-axis scoring: is this something I'd like, and is this something I need to know. + +The two questions are independent and get independent answers. Conflating them +is what made the old system's notifications wrong: its "importance" was +corroboration count, its relevance was three BM25 keywords, and its notification +lane consulted neither. + +Blocks are compared by z-score WITHIN block, never by raw cosine. Cosine +magnitude is not comparable across blocks -- "geopolitics" averages 0.501 +similarity against world-news text while "kentucky-louisville" averages 0.441 -- +so a plain max hands one block 78% of all items. See profile/CALIBRATION.md. +""" + +from __future__ import annotations + +import math +import sqlite3 +from datetime import datetime +from dataclasses import dataclass, field + +import numpy as np + +from . import cluster as cluster_mod +from . import store +from .cluster import Corroboration +from .embed import Embedder, embed_text +from .util import content_hash + + +@dataclass(frozen=True) +class BlockHit: + """The best-matching block of a kind, and how far above normal it scored.""" + + block: str + z: float + raw: float + + +@dataclass +class ClusterScore: + cluster_id: str + lead_title: str + last_seen: datetime | None + interest: BlockHit + importance: BlockHit + corroboration: Corroboration + velocity: float # distinct origins per hour since first seen + prominence: int # 1 if any member came from a front-page feed + ext_score: int | None # best aggregator score (HN points) in the cluster + veto: str | None + demote: str | None + demote_penalty: float + interest_score: float + importance_score: float + reasons: list[str] = field(default_factory=list) + + +class ProfileVectors: + """Embedded probes for every block in the profile. + + Probe vectors are cached in the same `embeddings` table as items, keyed by + content hash, so editing one probe re-embeds one probe rather than all of + them. + """ + + def __init__(self, con: sqlite3.Connection, profile: dict, embedder: Embedder): + self.blocks: dict[str, list[tuple[str, np.ndarray, dict]]] = {} + for kind in ("interest", "veto", "demote", "importance"): + entries = [] + for block in profile.get(kind, []): + pairs = [(content_hash(p), p) for p in block["probes"]] + vecs = embedder.encode_cached(con, pairs) + mat = np.stack([vecs[h] for h, _ in pairs]) + entries.append((block["id"], mat, block)) + self.blocks[kind] = entries + + def best(self, kind: str, vec: np.ndarray, weighted: bool = True) -> tuple[str, float]: + """Highest raw cosine across blocks of `kind`, as (block_id, cosine).""" + best_id, best = "", -1.0 + for block_id, mat, block in self.blocks[kind]: + raw = float((mat @ vec).max()) + if weighted: + raw *= block.get("weight", 1.0) + if raw > best: + best_id, best = block_id, raw + return best_id, best + + def raw_matrix(self, kind: str, vecs: np.ndarray) -> dict[str, np.ndarray]: + """Per-block raw cosine for every row of `vecs`, unweighted.""" + return {bid: (vecs @ mat.T).max(axis=1) for bid, mat, _ in self.blocks[kind]} + + def weight(self, kind: str, block_id: str) -> float: + for bid, _, block in self.blocks[kind]: + if bid == block_id: + return float(block.get("weight", 1.0)) + return 1.0 + + def block(self, kind: str, block_id: str) -> dict: + for bid, _, b in self.blocks[kind]: + if bid == block_id: + return b + return {} + + +def _z(values: np.ndarray) -> np.ndarray: + """Z-score, safe when a block is flat across the population.""" + std = float(values.std()) + if std < 1e-9: + return np.zeros_like(values) + return (values - float(values.mean())) / std + + +def score_clusters( + con: sqlite3.Connection, + profile_vectors: ProfileVectors, + scoring_cfg: dict, + clustering_cfg: dict, + since: str | None = None, +) -> list[ClusterScore]: + """Score every cluster in the window. + + The z-score population is the window itself, so scores are relative to + "what news looked like recently" rather than to an absolute cosine value + that means nothing on its own. + """ + rows = con.execute( + """SELECT c.id, c.centroid, c.size, c.first_seen_at, c.last_seen_at, + (SELECT title FROM items WHERE id = c.lead_item_id) AS lead + FROM clusters c + WHERE (? IS NULL OR c.last_seen_at >= ?) + ORDER BY c.last_seen_at DESC""", + (since, since), + ).fetchall() + if not rows: + return [] + + 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) + + interest_raw = profile_vectors.raw_matrix("interest", mat) + importance_raw = profile_vectors.raw_matrix("importance", mat) + veto_raw = profile_vectors.raw_matrix("veto", mat) + demote_raw = profile_vectors.raw_matrix("demote", mat) + + interest_z = {b: _z(v) * profile_vectors.weight("interest", b) for b, v in interest_raw.items()} + importance_z = { + b: _z(v) * profile_vectors.weight("importance", b) for b, v in importance_raw.items() + } + veto_z = {b: _z(v) for b, v in veto_raw.items()} + + def argbest(per_block: dict[str, np.ndarray], i: int) -> tuple[str, float]: + return max(((b, float(v[i])) for b, v in per_block.items()), key=lambda kv: kv[1]) + + out: list[ClusterScore] = [] + for i, row in enumerate(rows): + i_block, i_z = argbest(interest_z, i) + m_block, m_z = argbest(importance_z, i) + + # --- veto: absolute similarity OR a margin over the interest signal --- + veto_hit = None + best_veto_z = -9.0 + for b, v in veto_z.items(): + if veto_raw[b][i] >= scoring_cfg["veto_absolute"]: + veto_hit = b + if float(v[i]) > best_veto_z: + best_veto_z, best_veto_block = float(v[i]), b + if veto_hit is None and best_veto_z > i_z + scoring_cfg["veto_margin_over_interest"]: + veto_hit = best_veto_block + + # --- demote: a penalty, not an exclusion --- + demote_hit, penalty = None, 0.0 + for b, v in demote_raw.items(): + block = profile_vectors.block("demote", b) + if float(v[i]) >= scoring_cfg.get("demote_absolute", 0.55): + p = float(block.get("penalty", 1.0)) + if p > penalty: + demote_hit, penalty = b, p + + corr = cluster_mod.corroboration( + con, row["id"], clustering_cfg["syndication_threshold"], + scoring_cfg["corroboration_pickup_weight"], + scoring_cfg["corroboration_depth_weight"]) + + meta = con.execute( + """SELECT MAX(s.prominence) AS prom, MAX(i.ext_score) AS ext + 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 = ?""", + (row["id"],), + ).fetchone() + + # Velocity is a GROWTH rate, so it is only defined once a story has been + # picked up more than once. A singleton has no growth: computing + # 1 origin / 0.5h floor gave every single-source cluster a spurious + # velocity of 2.0, which is most of the corpus. + if corr.origins < 2: + velocity = 0.0 + else: + hours = max( + 0.5, + (cluster_mod._parse(row["last_seen_at"]) + - cluster_mod._parse(row["first_seen_at"])).total_seconds() / 3600.0, + ) + velocity = corr.origins / hours + + out.append( + ClusterScore( + cluster_id=row["id"], + lead_title=row["lead"] or "", + last_seen=cluster_mod._parse(row["last_seen_at"]), + interest=BlockHit(i_block, i_z, float(interest_raw[i_block][i])), + importance=BlockHit(m_block, m_z, float(importance_raw[m_block][i])), + corroboration=corr, + velocity=velocity, + prominence=int(meta["prom"] or 0), + ext_score=meta["ext"], + veto=veto_hit, + demote=demote_hit, + demote_penalty=penalty, + interest_score=i_z - penalty, + importance_score=_importance(m_z, corr, velocity, meta, scoring_cfg), + ) + ) + return out + + +def _importance( + block_z: float, corr: Corroboration, velocity: float, meta, cfg: dict +) -> float: + """Importance is NOT corroboration count. + + That was the old system's mistake: with a dozen general-news feeds all + reprinting the same wire, "how many sources carried it" measured the + subscription list rather than the news. Corroboration is one term among + several here, and burst velocity -- how fast independent newsrooms picked + it up -- carries as much weight, because 0 to 6 origins in ninety minutes is + what breaking news actually looks like. + """ + w = cfg.get("importance_weights", {}) + score = w.get("block", 1.0) * block_z + score += w.get("corroboration", 0.4) * math.log1p(max(0.0, corr.score)) + score += w.get("velocity", 0.5) * math.log1p(max(0.0, velocity)) + score += w.get("prominence", 0.3) * int(meta["prom"] or 0) + if meta["ext"]: + score += w.get("aggregator", 0.3) * math.log1p(float(meta["ext"]) / 100.0) + return score diff --git a/news/store.py b/news/store.py index f723d14..b80bd6b 100644 --- a/news/store.py +++ b/news/store.py @@ -80,6 +80,20 @@ CREATE TABLE IF NOT EXISTS cluster_members ( ); CREATE INDEX IF NOT EXISTS cluster_members_item ON cluster_members(item_id); +-- The suppression and budget ledger. Every send is recorded, including dry +-- runs, so `simulate` and the live router share one notion of "already sent". +CREATE TABLE IF NOT EXISTS notification_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + cluster_id TEXT NOT NULL, + lane TEXT NOT NULL, + score REAL, + origins_at_notify INTEGER, + dry_run INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL +); +CREATE INDEX IF NOT EXISTS notification_log_created ON notification_log(created_at); +CREATE INDEX IF NOT EXISTS notification_log_cluster ON notification_log(cluster_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, diff --git a/profile/config.toml b/profile/config.toml index ddd4228..d04c79c 100644 --- a/profile/config.toml +++ b/profile/config.toml @@ -67,3 +67,43 @@ veto_margin_over_interest = 0.3 # their own copy. Raise pickup_weight if you add local papers running wire. corroboration_pickup_weight = 0.3 corroboration_depth_weight = 0.2 + +[routing] +# Daily budgets. Deliberately small: the point of the system is a handful of +# well-chosen interruptions, not a feed. +interest_per_day = 3 +important_per_day = 2 +# Hard cap on waking the reader during quiet hours, inside important_per_day. +important_per_night = 1 + +# Thresholds are in z-score units -- "how many standard deviations above normal +# for this block", not raw cosine. Calibrate with `news simulate`, which reports +# the notifications-per-day these values would actually have produced. +interest_threshold = 2.0 +important_threshold = 2.0 +digest_floor = 0.5 +# A floor alone yielded 198 items a day. The digest is capped by count and fed +# strongest-first, so the cap is always spent on the best candidates. +digest_max_items = 25 +# Serendipity slots are filled from items BELOW this interest score, so they are +# genuinely outside the profile rather than near-misses. +serendipity_max_interest = 0.5 + +# Aggregator maturation: an HN post at 3 points is noise, at 300 it is news. +hn_min_score = 150 + +# Nothing older than this may generate a notification. Feeds return large +# archives on first fetch -- one run pulled 88 items going back five months -- +# and those belong in the scoring corpus but not in your pocket. +max_age_hours = 48 + +[scoring.importance_weights] +# Importance is NOT corroboration count -- that was the old system's error. +# Velocity is weighted as heavily as corroboration: 0 to 6 independent origins +# in ninety minutes is what breaking news looks like, and a story sitting at 6 +# origins for three days is not. +block = 1.0 +corroboration = 0.4 +velocity = 0.5 +prominence = 0.3 +aggregator = 0.3 diff --git a/profile/profile.toml b/profile/profile.toml index 39ea736..783d5b9 100644 --- a/profile/profile.toml +++ b/profile/profile.toml @@ -338,6 +338,21 @@ probes = [ "A missing person case takes a turn as investigators search a rural property.", ] +# Recurring columns, link round-ups and daily photo features. These are often +# genuinely on-topic -- Astronomy Picture of the Day matches space-aerospace +# correctly -- but a feature that publishes on a schedule is not news breaking, +# and should never be an interruption. Demoted rather than vetoed so they still +# reach the digest. +[[demote]] +id = "recurring-feature" +penalty = 1.5 +probes = [ + "Catching Our Eye: a round-up of links and short items we noticed this week.", + "Astronomy Picture of the Day: today's image, with a short explanatory caption.", + "Your morning briefing: everything you need to know today, in five bullet points.", + "This week in review: the stories we covered, collected.", +] + # ============================================= the importance lane =========== # Independent of interest. Vetoes still apply here -- v0.2's bypass lane ignored # relevance entirely (gate.rs:302), which let blocklisted items fire at priority 5. diff --git a/tests/test_route.py b/tests/test_route.py new file mode 100644 index 0000000..3d8ee27 --- /dev/null +++ b/tests/test_route.py @@ -0,0 +1,156 @@ +"""Routing decisions. + +The most important assertion here is that a veto blocks BOTH lanes. In the old +system the bypass lane never consulted relevance (gate.rs:302), so a blocklisted +item could still fire at priority 5 -- which is how a baseball recap could +interrupt someone who had blocklisted sports. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta + +import pytest + +from news import store +from news.cluster import Corroboration +from news.route import DIGEST, DROP, IMPORTANT, INTEREST, in_quiet_hours, route +from news.score import BlockHit, ClusterScore + +NOW = datetime(2026, 9, 8, 14, 0, tzinfo=UTC) # daytime +NIGHT = datetime(2026, 9, 8, 23, 30, tzinfo=UTC) # inside 22:00-07:00 + +ROUTING = { + "interest_per_day": 2, "important_per_day": 1, "important_per_night": 1, + "interest_threshold": 2.0, "important_threshold": 2.0, "digest_floor": 0.5, + "hn_min_score": 150, "max_age_hours": 48, "digest_max_items": 25, + "serendipity_max_interest": 0.5, +} +DELIVERY = { + "quiet_hours": {"start": "22:00", "end": "07:00"}, + "importance_may_wake": True, + "conflict_general_requires_importance": True, + "serendipity_slots": 0, +} + + +def mk(cid, *, interest=0.0, importance=0.0, veto=None, block="urbanism", + imp_block="war-escalation", ext=None, origins=2, last_seen=NOW): + return ClusterScore( + cluster_id=cid, lead_title=cid, last_seen=last_seen, + interest=BlockHit(block, interest, 0.5), + importance=BlockHit(imp_block, importance, 0.5), + corroboration=Corroboration(origins, 0, 0, float(origins)), + velocity=0.0, prominence=0, ext_score=ext, veto=veto, demote=None, + demote_penalty=0.0, interest_score=interest, importance_score=importance) + + +@pytest.fixture +def con(tmp_path): + return store.connect(tmp_path / "r.db") + + +def lanes(decisions): + return {d.cluster_id: d.lane for d in decisions} + + +def test_veto_blocks_the_importance_lane_too(con): + """The v0.2 bug: bypass ignored relevance, so vetoed items still fired.""" + out = route(con, [mk("v", interest=5.0, importance=9.0, veto="us-big-three-sports")], + ROUTING, DELIVERY, NOW) + assert lanes(out)["v"] == DROP + assert out[0].reason.startswith("veto:") + + +def test_high_interest_alerts_during_the_day(con): + out = route(con, [mk("a", interest=3.0)], ROUTING, DELIVERY, NOW) + assert lanes(out)["a"] == INTEREST + + +def test_interest_never_overrides_quiet_hours(con): + out = route(con, [mk("a", interest=9.0)], ROUTING, DELIVERY, NIGHT) + assert lanes(out)["a"] == DIGEST + assert out[0].reason == "quiet-hours" + + +def test_importance_may_wake_but_only_within_the_nightly_cap(con): + scores = [mk("i1", importance=9.0), mk("i2", importance=8.0)] + out = lanes(route(con, scores, ROUTING, DELIVERY, NIGHT)) + assert out["i1"] == IMPORTANT, "the top importance item may wake" + assert out["i2"] == DIGEST, "the nightly cap is 1" + + +def test_importance_cannot_wake_when_disabled(con): + delivery = {**DELIVERY, "importance_may_wake": False} + out = route(con, [mk("i", importance=9.0)], ROUTING, delivery, NIGHT) + assert lanes(out)["i"] == DIGEST + + +def test_daily_budget_is_spent_on_the_strongest_candidates(con): + """Budget must go to the best items, not to whichever arrived first.""" + scores = [mk("weak", interest=2.1), mk("strong", interest=8.0), mk("mid", interest=4.0)] + out = lanes(route(con, scores, {**ROUTING, "interest_per_day": 1}, DELIVERY, NOW)) + assert out["strong"] == INTEREST + assert out["weak"] == DIGEST and out["mid"] == DIGEST + + +def test_stale_clusters_never_notify(con): + old = NOW - timedelta(days=5) + out = route(con, [mk("s", interest=9.0, last_seen=old)], ROUTING, DELIVERY, NOW) + assert lanes(out)["s"] == DROP + assert out[0].reason == "stale" + + +def test_immature_aggregator_items_are_held(con): + """An HN post at 3 points is noise; it is re-scored as its score climbs.""" + out = route(con, [mk("hn", interest=9.0, ext=5, origins=1)], ROUTING, DELIVERY, NOW) + assert lanes(out)["hn"] == DROP + assert out[0].reason == "aggregator-immature" + ripe = route(con, [mk("hn", interest=9.0, ext=400, origins=1)], ROUTING, DELIVERY, NOW) + assert lanes(ripe)["hn"] == INTEREST + + +def test_general_conflict_needs_importance_to_interrupt(con): + """Connor: war interest is low, but all wars are important.""" + routine = mk("c", interest=9.0, importance=0.1, block="conflict-general") + assert lanes(route(con, [routine], ROUTING, DELIVERY, NOW))["c"] == DIGEST + major = mk("c", interest=9.0, importance=5.0, block="conflict-general") + assert lanes(route(con, [major], ROUTING, DELIVERY, NOW))["c"] == IMPORTANT + + +def test_digest_is_capped_by_count(con): + scores = [mk(f"d{i}", interest=1.0) for i in range(40)] + out = lanes(route(con, scores, {**ROUTING, "digest_max_items": 5}, DELIVERY, NOW)) + assert sum(1 for v in out.values() if v == DIGEST) == 5 + + +def test_serendipity_fills_from_outside_the_profile(con): + delivery = {**DELIVERY, "serendipity_slots": 2} + # On-profile items must outrank the off-profile ones and fill the cap, + # otherwise the off-profile items make the digest on merit and the reserved + # slots have nothing to do. (An earlier version of this test got that + # backwards and failed against correct code.) + scores = [mk(f"on{i}", interest=1.9, importance=1.9) for i in range(30)] + scores += [mk("off1", interest=-1.0, importance=1.0), + mk("off2", interest=-2.0, importance=0.9)] + out = route(con, scores, {**ROUTING, "digest_max_items": 5}, delivery, NOW) + picked = {d.cluster_id for d in out if d.reason == "serendipity"} + assert picked == {"off1", "off2"} + + +def test_already_notified_clusters_are_not_repeated(con): + con.execute( + "INSERT INTO notification_log (cluster_id,lane,score,created_at) VALUES (?,?,?,?)", + ("a", INTEREST, 5.0, NOW.isoformat())) + out = route(con, [mk("a", interest=9.0)], ROUTING, DELIVERY, NOW) + assert lanes(out)["a"] == DROP + assert out[0].reason == "already-notified" + + +@pytest.mark.parametrize("hhmm,expected", [ + ("23:30", True), ("02:00", True), ("06:59", True), + ("07:00", False), ("14:00", False), ("21:59", False), ("22:00", True)]) +def test_quiet_hours_wrap_midnight(hhmm, expected): + h, m = map(int, hhmm.split(":")) + now = datetime(2026, 9, 8, h, m, tzinfo=UTC) + assert in_quiet_hours(now, {"start": "22:00", "end": "07:00"}) is expected