228 lines
7.3 KiB
Python
228 lines
7.3 KiB
Python
"""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)
|