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()