apodex/frontier-agent-demo

1"""In-process negative cache for web scrape URLs."""2 3from __future__ import annotations4 5import asyncio6import os7import threading8import time9from collections.abc import Awaitable, Callable10from dataclasses import dataclass, field11 12from frontier_agent.infra.usage_meter import record_api_request13 14# ── Tunables ──────────────────────────────────────────────────────────────15 16# Ban durations per status (seconds).17_BAN_403 = 3600           # 1 hour — matches Jina's rolling window.18_BAN_422 = 1800           # 30 min — paywall/SPA/empty content.19_BAN_429 = 300            # 5 min — transient rate limit.20 21# Minimum consecutive failures before a 422 results in a ban.22# 403 and 429 ban on the first occurrence.23_MIN_FAILS_422 = 224 25# Status codes we track. Everything else is not cached.26_TRACKED_STATUSES = {403, 422, 429}27 28 29@dataclass30class _Entry:31    status: int32    fail_count: int = 033    ban_until: float = 0.0      # unix time; 0 means not banned34    last_failed_at: float = field(default_factory=time.time)35 36 37class _ScrapeCache:38    """Thread-safe negative cache for scrape URLs."""39 40    def __init__(self) -> None:41        self._data: dict[str, _Entry] = {}42        self._lock = threading.Lock()43 44    def check(self, url: str, now: float | None = None) -> _Entry | None:45        """Return active entry if URL is currently banned, else None."""46        if not url:47            return None48        t = now if now is not None else time.time()49        with self._lock:50            entry = self._data.get(url)51            if entry is None:52                return None53            if entry.ban_until > t:54                return entry55            # Expired — drop it so future failures start fresh.56            if entry.ban_until and entry.ban_until <= t:57                self._data.pop(url, None)58            return None59 60    def record_failure(self, url: str, status: int, now: float | None = None) -> _Entry | None:61        """Record a failure; set ban_until if rules trigger. Returns updated entry."""62        if not url or status not in _TRACKED_STATUSES:63            return None64        t = now if now is not None else time.time()65        with self._lock:66            entry = self._data.get(url)67            if entry is None or entry.status != status:68                # Reset counter if status changed (e.g., 429 → 403).69                entry = _Entry(status=status, fail_count=0, last_failed_at=t)70            entry.fail_count += 171            entry.last_failed_at = t72            entry.status = status73 74            if status == 403:75                entry.ban_until = t + _BAN_40376            elif status == 429:77                entry.ban_until = t + _BAN_42978            elif status == 422 and entry.fail_count >= _MIN_FAILS_422:79                entry.ban_until = t + _BAN_42280 81            self._data[url] = entry82            return entry83 84    def record_success(self, url: str) -> None:85        """Clear any prior failure record for this URL."""86        if not url:87            return88        with self._lock:89            self._data.pop(url, None)90 91    def clear(self) -> None:92        """Drop all entries. Primarily for tests."""93        with self._lock:94            self._data.clear()95 96    def size(self) -> int:97        with self._lock:98            return len(self._data)99 100 101# Module-level singleton. Callers should import this directly.102cache = _ScrapeCache()103 104 105# ── Positive scrape cache (cross-run, single-flight) ───────────────────────106#107# The negative cache above only suppresses re-hammering KNOWN-BAD URLs. This108# positive cache stores SUCCESSFUL scrape content so the same URL fetched by109# many sibling agents costs one Jina round-trip, not N.110#111# Why it pays: sibling agents researching one question converge on the same112# canonical pages, so unique URLs run far below total fetches (typically a113# 3-5x ratio on a fan-out run). Agents dispatched with asyncio.gather share114# one process and event loop, so a module-level singleton reaches all of them115# with zero plumbing.116#117# Scope = SCRAPE ONLY. The per-call SUMMARY_LLM extraction is deliberately NOT118# cached: sharing raw page bytes can't homogenise the runs' reasoning (same URL119# is the same content for everyone), but sharing extraction would leak one120# run's info_to_extract focus into another and couple the otherwise-independent121# trajectories. Successes only — a failed scrape is never stored, and a waiter122# whose leader failed/timed-out falls back to its own fetch, so the cache never123# converts one transient failure into a correlated N-run failure.124 125# Follower wait cap before falling back to an independent scrape. Bounds the126# "leader stalled → everyone blocks" failure mode.127_SINGLE_FLIGHT_TIMEOUT_S = 90.0128# Cap on distinct cached pages (FIFO eviction) — guards memory on long runs.129_MAX_CACHE_ENTRIES = 1024130 131 132def _positive_cache_enabled() -> bool:133    """On by default; set ``WEB_FETCH_SCRAPE_CACHE=0`` (or false/off) to disable."""134    raw = (os.environ.get("WEB_FETCH_SCRAPE_CACHE") or "").strip().lower()135    return raw not in {"0", "false", "off", "no"}136 137 138class ScrapeUnavailable(Exception):139    """Raised by a scrape callable when content could not be obtained.140 141    Signals the cache to NOT store a result and lets waiters fall back to an142    independent fetch. Carries the original error string for the caller's143    user-facing message.144    """145 146 147class ScrapeResultCache:148    """Process-global positive cache for successful scrapes, with single-flight.149 150    ``get_or_scrape(url, scrape_fn)`` returns cached content on a hit; on a miss151    the first caller (the *leader*) runs ``scrape_fn`` while concurrent callers152    for the same URL (*followers*) await its result instead of duplicating the153    fetch. The leader caches only successful, non-empty content.154    """155 156    def __init__(self) -> None:157        self._content: dict[str, str] = {}158        self._inflight: dict[str, asyncio.Future[str]] = {}159        self._lock = asyncio.Lock()160        self.hits = 0161        self.misses = 0162        self.coalesced = 0163 164    async def get_or_scrape(165        self,166        url: str,167        scrape_fn: Callable[[], Awaitable[str]],168        *,169        single_flight_timeout: float = _SINGLE_FLIGHT_TIMEOUT_S,170        should_cache: Callable[[str], bool] | None = None,171    ) -> str:172        """Return cached content for ``url`` or run ``scrape_fn`` to produce it.173 174        ``scrape_fn`` must return the scraped content string on success and175        raise :class:`ScrapeUnavailable` (or any exception) on failure — only176        successful, non-empty returns accepted by ``should_cache`` are cached.177        ``should_cache`` defaults to accepting every non-empty string; callers178        can return a low-confidence result to the current request without179        poisoning later requests with it. Raises whatever ``scrape_fn`` raises180        for the leader; followers fall back to their own ``scrape_fn`` on the181        leader's failure or a wait timeout.182        """183        if not url or not _positive_cache_enabled():184            return await scrape_fn()185 186        async with self._lock:187            if url in self._content:188                self.hits += 1189                # A cache hit is one Jina round-trip saved.190                record_api_request("jina", requests=0, cache_hits=1)191                return self._content[url]192            fut = self._inflight.get(url)193            leader = fut is None194            if leader:195                self.misses += 1196                fut = asyncio.get_event_loop().create_future()197                self._inflight[url] = fut198            else:199                self.coalesced += 1200                # Coalesced follower shares the leader's round-trip —201                # also a saved upstream request.202                record_api_request("jina", requests=0, cache_hits=1)203 204        if not leader:205            try:206                # ``shield`` so a waiter's timeout can't cancel the shared work.207                return await asyncio.wait_for(208                    asyncio.shield(fut), timeout=single_flight_timeout,209                )210            except Exception:211                # Leader failed / timed out → independent fetch (resilience).212                return await scrape_fn()213 214        # Leader path.215        try:216            content = await scrape_fn()217        except BaseException as exc:218            async with self._lock:219                self._inflight.pop(url, None)220            if not fut.done():221                fut.set_exception(exc)222                # Retrieve eagerly so a leader failure with no waiting follower223                # doesn't log "Future exception was never retrieved". Waiters224                # (if any) still see it via their own ``await``.225                fut.exception()226            raise227        cacheable = isinstance(content, str) and bool(content.strip())228        if cacheable and should_cache is not None:229            try:230                cacheable = bool(should_cache(content))231            except Exception:232                # Cache policy is an optimization boundary: a buggy predicate233                # must not fail a successful scrape or strand coalesced waiters.234                cacheable = False235        async with self._lock:236            self._inflight.pop(url, None)237            if cacheable:238                if len(self._content) >= _MAX_CACHE_ENTRIES:239                    self._content.pop(next(iter(self._content)), None)240                self._content[url] = content241        if not fut.done():242            fut.set_result(content)243        return content244 245    def clear(self) -> None:246        """Drop all entries + counters. Primarily for tests."""247        self._content.clear()248        self._inflight.clear()249        self.hits = self.misses = self.coalesced = 0250 251    def stats(self) -> dict[str, int]:252        return {253            "hits": self.hits,254            "misses": self.misses,255            "coalesced": self.coalesced,256            "size": len(self._content),257        }258 259 260# Module-level singleton — shared across all sibling agents in the process.261scrape_result_cache = ScrapeResultCache()262 263 264def format_skip_message(url: str, entry: _Entry, now: float | None = None) -> str:265    """Format a short, LLM-facing message explaining why the URL was skipped."""266    t = now if now is not None else time.time()267    remaining = max(0, int(entry.ban_until - t))268    mins = remaining // 60269    reason = {270        403: "returned 403 (origin blocked or Jina URL ban)",271        422: f"returned 422 {entry.fail_count}x (paywall, empty, or unparseable content)",272        429: "was rate-limited (429)",273    }.get(entry.status, f"failed with status {entry.status}")274    return (275        f"URL skipped: {url} {reason} earlier in this session. "276        f"Cached for ~{mins} more min. Try a different source or search query."277    )278