"""Breaking-news watcher — Pilon 9 of the cache freshness defense. Polls a small set of high-credibility breaking-news RSS feeds every 5 min. For each new item, runs the volatility classifier (LLM) to identify the affected topics + entities, then triggers ``/v1/cache/invalidate`` on those topics so cached verdicts that depend on the just-changed reality are flushed within minutes of the breaking story going live. End-to-end latency target: under 30s from RSS publish to cache invalidation (network + LLM ≈ 5-15s in practice with local Qwen). Memory: keeps an in-process LRU of recently-seen URLs to avoid re-classifying the same article on every poll. Restart-safe: brain's ingest dedup also skips already-known URLs. """ from __future__ import annotations import asyncio import os from collections import deque from datetime import datetime, timezone from brain_api.services.classifier import ( ClaimVolatility, classify_claim_volatility, ) from brain_api.services.fact_status import canonicalize_triple from scheduler.brain_client import BrainClient from scheduler.config import ( DEFAULT_BREAKING_FEEDS, parse_csv, settings, ) from scheduler.feed_parser import ( FeedItem, dedup_by_url, fetch_feeds, filter_by_age, ) from shared.llm_client import LlmClient from shared.logging import get_logger log = get_logger(__name__) # In-process LRU of seen URLs — capped so restart doesn't accumulate forever. SEEN_LRU_MAX = 2000 _seen_urls: deque[str] = deque(maxlen=SEEN_LRU_MAX) _seen_set: set[str] = set() def _mark_seen(url: str) -> None: """Track URL as seen, evicting oldest if at capacity.""" if url in _seen_set: return if len(_seen_urls) == SEEN_LRU_MAX: # deque.append at maxlen drops the oldest; reflect in the set. oldest = _seen_urls[0] _seen_set.discard(oldest) _seen_urls.append(url) _seen_set.add(url) def _is_seen(url: str) -> bool: return url in _seen_set def _build_invalidation_targets( classification: ClaimVolatility, ) -> tuple[list[str], list[str]]: """Decide what to invalidate based on classifier output. Only volatile and evolving classifications trigger invalidation — stable items (background pieces, historical recap) shouldn't flush anything. Returns ``(topic_codes, entity_canonicals)``: - topic_codes: pass through directly to /v1/cache/invalidate - entity_canonicals: canonicalize each binding so PG can match against cached entity_bindings JSONB """ if classification.volatility not in ("volatile", "evolving"): return [], [] topics = list(classification.topic_codes or []) canonicals: list[str] = [] for b in classification.entity_bindings or []: if b.confidence < 0.6: # Low-confidence extractions are noisy — skip to avoid # accidental mass invalidation. continue canonicals.append( canonicalize_triple(b.subject, b.predicate, b.obj) ) return topics, canonicals async def _process_one_item( *, item: FeedItem, brain: BrainClient, llm: LlmClient, ) -> str: """Classify one breaking item and trigger invalidation if applicable. Returns a short label for telemetry: 'classified_no_action' / 'invalidated' / 'ignored_low_confidence' / 'classifier_failed'. """ # Title + summary is what we feed the classifier — full article body # would be expensive and the headline carries the signal we need. text = (item.title or "") if item.summary: text = f"{text}. {item.summary}" if not text.strip(): return "no_text" classification = await classify_claim_volatility(llm, claim=text) if classification.degraded: return "classifier_failed" topics, canonicals = _build_invalidation_targets(classification) if not topics and not canonicals: return "classified_no_action" # Dry-run first to count, then real invalidate. We tolerate partial # successes — if a network blip kills the real call, the next watcher # cycle will catch the same item again. result = await brain.invalidate( topic_codes=topics or None, entity_canonicals=canonicals or None, # Only invalidate verdicts written before this breaking story — # avoids racing with concurrent writes that may have used fresh # information already. since_iso=None, invalidate_gold=False, actor=f"breaking_watcher:{item.publisher}", reason=( f"breaking story: {item.title[:120]} " f"(vol={classification.volatility})" ), ) if result is None: return "invalidate_http_error" log.info( "watcher_invalidated", title=item.title[:80], publisher=item.publisher, volatility=classification.volatility, topics=topics, entities=len(canonicals), atoms=result.get("invalidated_atoms"), vcache=result.get("invalidated_vcache"), ) return "invalidated" async def _run_one_cycle(brain: BrainClient, llm: LlmClient) -> dict[str, int]: """Pull breaking feeds, classify novel items, invalidate as needed.""" feed_urls = parse_csv( os.environ.get("FEED_BREAKING_URLS"), DEFAULT_BREAKING_FEEDS ) if not feed_urls: return {"checked": 0, "novel": 0} items = await fetch_feeds(feed_urls) items = filter_by_age( items, min_age_s=settings.watcher_min_published_age_s, max_age_s=settings.watcher_max_published_age_s, ) items = dedup_by_url(items) # Drop items already processed in a previous cycle. novel = [i for i in items if not _is_seen(i.url)] novel.sort(key=lambda i: i.published_at, reverse=True) novel = novel[: settings.watcher_max_items_per_run] if not novel: return {"checked": len(items), "novel": 0} results: dict[str, int] = {} for item in novel: try: label = await _process_one_item( item=item, brain=brain, llm=llm ) results[label] = results.get(label, 0) + 1 except Exception as e: # noqa: BLE001 log.warning( "watcher_item_failed", url=item.url, error=f"{type(e).__name__}:{e}", ) results["item_error"] = results.get("item_error", 0) + 1 finally: _mark_seen(item.url) log.info( "watcher_cycle_done", feeds=len(feed_urls), items_total=len(items), novel=len(novel), breakdown=results, ) return {"checked": len(items), "novel": len(novel), **results} async def run_watcher(brain: BrainClient, llm: LlmClient) -> None: """Long-running task — polls breaking-news feeds at watcher_poll_interval_s.""" if not settings.watcher_enabled: log.info("watcher_disabled") return log.info( "watcher_loop_start", interval_s=settings.watcher_poll_interval_s, feeds=len( parse_csv( os.environ.get("FEED_BREAKING_URLS"), DEFAULT_BREAKING_FEEDS ) ), ) while True: try: await _run_one_cycle(brain, llm) except Exception: # noqa: BLE001 log.exception("watcher_cycle_error") await asyncio.sleep(settings.watcher_poll_interval_s)