From d05348d1e42cdc77ea96a8ac388b2153eb8dbbab Mon Sep 17 00:00:00 2001 From: Harwinder Sekhon Date: Tue, 8 Sep 2026 13:04:05 -0400 Subject: [PATCH] news archive: fix three bugs found reviewing 301bf90 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. A backfill pass silently wiped teacher identity from live rows. The provenance compare is `>=`, and backfill rows are llm-scored but carry no teacher_model/teacher_prompt_sha1 (neither source DB recorded one), so a bare assignment blanked what the live hook had stored. Re-running the backfill — e.g. after importing the live server's signals.db, which is the next planned step — would have erased it from every live row. COALESCE now keeps a stored identity when the incoming one is NULL, while a genuinely newer teacher still replaces it. Timestamps clamp for the same reason: backfill stamps historical values, so first_seen only moves earlier and last_seen only later. 2. A retried batch could be applied twice. executemany does not roll itself back, so a batch that half-landed and then hit a lock was re-applied whole by _with_lock_retry, double-counting seen_count on the rows that already landed. Writes now run in an explicit BEGIN IMMEDIATE with a rollback, which also takes the write lock up front so contention fails before any row is applied. 3. The archive hook could break a news fetch. archive_headlines swallows its own failures, but its ARGUMENTS are evaluated first, so a raising get_llm_client() escaped that guard. Guarded at the call site too: archiving is a side effect and must never cost us the headlines we just fetched. Both 1 and 2 were verified to reproduce against 301bf90 before fixing (teacher_model -> None; seen_count 3 instead of 2 on the rows that landed pre-failure) and are now regression tests. Also fixed two weak tests of my own: the disabled-by-config test never pointed NEWS_ARCHIVE_PATH at its temp dir, so its assertion passed regardless of behaviour and the write went to the real archive; and the neutral-headline test asserted the headline was archived without showing it is dropped downstream, which is the entire reason the hook sits where it does. Positional binding of the provenance rank is now asserted rather than trusted, since a miscount would shift every parameter silently. 483 -> 486 tests. Claude-Session: https://claude.ai/code/session_01Bzv4ETVapFNauFDQht7QtA --- news_archive.py | 48 ++++++++++++-- news_engine.py | 15 +++-- test_all.py | 169 ++++++++++++++++++++++++++++++++++++++++++++++-- 3 files changed, 216 insertions(+), 16 deletions(-) diff --git a/news_archive.py b/news_archive.py index 1965e1c..9322e03 100644 --- a/news_archive.py +++ b/news_archive.py @@ -205,7 +205,11 @@ class NewsArchive: " teacher_prompt_sha1, origin" ") VALUES (?,?,?,?,?,?,?,?,?,1,?,?,?,?,?,?,?,?)" " ON CONFLICT(hash) DO UPDATE SET" - " last_seen = excluded.last_seen," + # The backfill stamps historical timestamps, so clamp rather + # than assign: first_seen only ever moves earlier, last_seen + # only ever moves later. + " first_seen = MIN(headlines.first_seen, excluded.first_seen)," + " last_seen = MAX(headlines.last_seen, excluded.last_seen)," " seen_count = headlines.seen_count + 1," # Upgrade-only: a weaker scorer never overwrites a stronger # one's label, but re-seeing a headline always counts. @@ -218,10 +222,17 @@ class NewsArchive: " ELSE headlines.currencies_json END," f" keywords_json = CASE WHEN {wins} THEN excluded.keywords_json" " ELSE headlines.keywords_json END," - f" teacher_model = CASE WHEN {wins} THEN excluded.teacher_model" + # COALESCE, not a bare overwrite: the backfill re-ingests + # llm-scored headlines with NO teacher identity (neither source + # DB recorded one), and rank >= rank is true, so a plain + # assignment blanks what the live hook already stored. + f" teacher_model = CASE WHEN {wins}" + " THEN COALESCE(excluded.teacher_model," + " headlines.teacher_model)" " ELSE headlines.teacher_model END," f" teacher_prompt_sha1 = CASE WHEN {wins}" - " THEN excluded.teacher_prompt_sha1" + " THEN COALESCE(excluded.teacher_prompt_sha1," + " headlines.teacher_prompt_sha1)" " ELSE headlines.teacher_prompt_sha1 END," f" score_source = CASE WHEN {wins} THEN excluded.score_source" " ELSE headlines.score_source END," @@ -233,6 +244,15 @@ class NewsArchive: " published_at = COALESCE(excluded.published_at," " headlines.published_at)") + # The rank is bound once per CASE arm. Positional binding means a + # miscount silently shifts every parameter, so assert it instead of + # trusting the arithmetic: 17 INSERT values + one rank per arm. + n_rank = sql.count("? >= ") + n_values = 17 + assert sql.count("?") == n_values + n_rank, ( + f"placeholder count drifted: {sql.count('?')} " + f"!= {n_values} + {n_rank}") + params = [] for row in rows: rank = _SOURCE_RANK.get(row["score_source"], 0) @@ -244,11 +264,27 @@ class NewsArchive: row["currencies_json"], row["keywords_json"], row["score_source"], row["teacher_model"], row["teacher_prompt_sha1"], row.get("origin", "live"), - ) + (rank,) * sql.count("? >=")) + ) + (rank,) * n_rank) + + def _write() -> None: + # BEGIN IMMEDIATE takes the write lock up front, so contention + # fails here — before any row is applied — and the rollback + # makes the batch atomic if it fails later. Without this a + # partial executemany would be re-applied by the retry and + # double-count seen_count on the rows that already landed. + self._conn.execute("BEGIN IMMEDIATE") + try: + self._conn.executemany(sql, params) + self._conn.commit() + except Exception: + try: + self._conn.rollback() + except Exception: + pass + raise with self._lock: - self._with_lock_retry(lambda: self._conn.executemany(sql, params)) - self._conn.commit() + self._with_lock_retry(_write) return len(rows) diff --git a/news_engine.py b/news_engine.py index 60f173b..2c6ece4 100755 --- a/news_engine.py +++ b/news_engine.py @@ -921,10 +921,17 @@ class NewsEngine: # corpus built from what the context carries holds almost no # genuinely neutral headlines and would teach a trained scorer # that "neutral" barely exists. Never raises. - news_archive.archive_headlines( - headlines, - teacher_model=getattr(get_llm_client(), "_model", None), - teacher_prompt_sha1=_LLM_SENTIMENT_PROMPT_SHA1) + # Guarded at the call site as well as inside archive_headlines: + # arguments are evaluated BEFORE that function's own try/except, so + # a raising get_llm_client() would escape it. Archiving is a side + # effect — it must never cost us the headlines we just fetched. + try: + news_archive.archive_headlines( + headlines, + teacher_model=getattr(get_llm_client(), "_model", None), + teacher_prompt_sha1=_LLM_SENTIMENT_PROMPT_SHA1) + except Exception as e: + logger.debug(f"news archive skipped, headlines unaffected: {e}") return headlines diff --git a/test_all.py b/test_all.py index c4422ef..db380c0 100644 --- a/test_all.py +++ b/test_all.py @@ -10860,6 +10860,98 @@ def _(): arch.close() +@test("news archive: a backfill pass never blanks a live row's teacher identity") +def _(): + """The backfill re-ingests llm-scored headlines carrying NO teacher + identity (neither source DB recorded one) and rank >= rank is true, so + a bare assignment wipes what the live hook stored.""" + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "newsdata.db") + arch = _na.NewsArchive(path) + try: + arch.store([_na.build_row( + _m32_head(source="llm"), teacher_model="deepseek-v4-flash", + teacher_prompt_sha1="d32e04c6", origin="live", + first_seen="2026-09-08T10:00:00+00:00", + last_seen="2026-09-08T10:00:00+00:00")]) + arch.store([_na.build_row( + _m32_head(source="llm"), teacher_model=None, + teacher_prompt_sha1=None, origin="backfill", + first_seen="2026-08-13T09:00:00+00:00", + last_seen="2026-08-13T09:00:00+00:00")]) + con = sqlite3.connect(f"file:{path}?mode=ro", uri=True) + try: + model, sha, first, last = con.execute( + "SELECT teacher_model, teacher_prompt_sha1, first_seen," + " last_seen FROM headlines").fetchone() + finally: + con.close() + assert model == "deepseek-v4-flash", f"teacher identity wiped: {model}" + assert sha == "d32e04c6", sha + # historical timestamps clamp outward, never overwrite + assert first.startswith("2026-08-13"), f"first_seen: {first}" + assert last.startswith("2026-09-08"), f"last_seen: {last}" + + # A genuinely newer teacher still replaces the old identity. + arch.store([_na.build_row( + _m32_head(source="llm"), teacher_model="deepseek-v5", + teacher_prompt_sha1="newsha")]) + con = sqlite3.connect(f"file:{path}?mode=ro", uri=True) + try: + assert con.execute( + "SELECT teacher_model FROM headlines").fetchone()[0] \ + == "deepseek-v5" + finally: + con.close() + finally: + arch.close() + + +@test("news archive: a retried batch is applied exactly once, never twice") +def _(): + """executemany does not roll itself back. Without an explicit + transaction a batch that half-lands and then hits a lock is re-applied + whole by the retry, double-counting the rows that already landed.""" + class _Flaky: + """Real connection, but the first executemany lets 2 of 5 rows land + and then raises 'database is locked'.""" + + def __init__(self, con): + self._c = con + self.attempts = 0 + + def __getattr__(self, name): + return getattr(self._c, name) + + def executemany(self, sql, params): + self.attempts += 1 + if self.attempts == 1: + self._c.executemany(sql, list(params)[:2]) + raise sqlite3.OperationalError("database is locked") + return self._c.executemany(sql, params) + + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "newsdata.db") + arch = _na.NewsArchive(path) + try: + rows = [_na.build_row(_m32_head(headline=f"h{i}")) + for i in range(5)] + arch.store(rows) # seed: seen_count == 1 + arch._conn = _Flaky(arch._conn) + arch.store(rows) # half-lands, retries + assert arch._conn.attempts == 2, arch._conn.attempts + finally: + arch.close() + con = sqlite3.connect(f"file:{path}?mode=ro", uri=True) + try: + counts = dict(con.execute( + "SELECT headline, seen_count FROM headlines")) + finally: + con.close() + doubled = {k: v for k, v in counts.items() if v != 2} + assert not doubled, f"retry double-applied: {doubled}" + + @test("news archive: concurrent writers lose no rows and no seen_count") def _(): """The bot, the dashboard publisher and the tests all open this file at @@ -10925,17 +11017,34 @@ def _(): import app_config as _ac with tempfile.TemporaryDirectory() as d: path = os.path.join(d, "newsdata.db") - saved = os.environ.get("NEWS_ARCHIVE_ENABLED") + saved_en = os.environ.get("NEWS_ARCHIVE_ENABLED") + saved_path = os.environ.get("NEWS_ARCHIVE_PATH") + saved_archives = dict(_na._archives) + _na._archives.clear() + # The path MUST be pointed at the temp dir, or the assertion below + # is vacuous: the write would land in the default logs/newsdata.db. + os.environ["NEWS_ARCHIVE_PATH"] = path os.environ["NEWS_ARCHIVE_ENABLED"] = "0" try: assert not _ac.news_archive_enabled() _na.archive_headlines([_m32_head()]) + assert not os.path.exists(path), "nothing may be written when off" + # ...and the same call with it on DOES write, proving the check + # above is testing the switch and not a typo in the path. + os.environ["NEWS_ARCHIVE_ENABLED"] = "1" + _na.archive_headlines([_m32_head()]) + assert os.path.exists(path), "the switch is what gates the write" finally: - if saved is None: - os.environ.pop("NEWS_ARCHIVE_ENABLED", None) - else: - os.environ["NEWS_ARCHIVE_ENABLED"] = saved - assert not os.path.exists(path), "nothing may be written when off" + for arch in _na._archives.values(): + arch.close() + _na._archives.clear() + _na._archives.update(saved_archives) + for key, val in (("NEWS_ARCHIVE_ENABLED", saved_en), + ("NEWS_ARCHIVE_PATH", saved_path)): + if val is None: + os.environ.pop(key, None) + else: + os.environ[key] = val @test("news archive: a write failure never raises into a scan") @@ -11076,6 +11185,11 @@ def _(): rows = _m32_rows(path) assert "Gold steady as traders await US CPI" in rows, \ "the neutral headline must still be archived" + # ...and it really is dropped downstream — the filter + # get_fundamental_context applies before building the context. + kept = [h for h in heads if not _is_neutral_headline(h)] + assert kept == [], \ + "the context keeps it, so archiving downstream would be fine" finally: NewsEngine._source_health = saved_health for arch in _na._archives.values(): @@ -11088,6 +11202,49 @@ def _(): os.environ["NEWS_ARCHIVE_PATH"] = saved_env +@test("news archive: the hook cannot break a news fetch, whatever it hits") +def _(): + """archive_headlines swallows its own failures, but its ARGUMENTS are + evaluated first — a raising get_llm_client() would escape that guard.""" + import news_engine as _ne_mod + saved_client = _ne_mod.get_llm_client + saved_archive = _na.archive_headlines + saved_health = NewsEngine._source_health + NewsEngine._source_health = {} + + def _boom_client(): + raise RuntimeError("config exploded") + + def _boom_archive(*a, **k): + raise RuntimeError("archive exploded") + + rss = (b"" + b"Gold surges as Fed signals rate cuts" + b"" + _m19_pubdate(1) + b"" + b"") + + class _Sess: + def get(self, url, timeout=None): + return type("R", (), {"status_code": 200, "content": rss})() + + try: + for label, patch in (("client", "client"), ("archive", "archive")): + _ne_mod.get_llm_client = (_boom_client if patch == "client" + else saved_client) + _na.archive_headlines = (_boom_archive if patch == "archive" + else saved_archive) + cfg = NewsConfig() + cfg.rss_feeds = ["https://feed.example/rss"] + ne = NewsEngine(cfg) + ne._session = _Sess() + heads = ne._fetch_news_sentiment_raw() # must NOT raise + assert len(heads) == 1, f"{label}: fetch lost its headlines" + finally: + _ne_mod.get_llm_client = saved_client + _na.archive_headlines = saved_archive + NewsEngine._source_health = saved_health + + @test("news archive: the backtester never builds a NewsEngine, so replays never archive") def _(): src = open("backtest/gpu_backtest.py").read()