news archive: fix three bugs found reviewing 301bf90

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.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Bzv4ETVapFNauFDQht7QtA

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._with_lock_retry(_write)
-            self._conn.commit()
         return len(rows)
 
 

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.
+        # 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
 

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()])
-        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"
+            # ...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:
+            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"<rss><channel>"
+           b"<item><title>Gold surges as Fed signals rate cuts</title>"
+           b"<pubDate>" + _m19_pubdate(1) + b"</pubDate></item>"
+           b"</channel></rss>")
+
+    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()