Cache Tavily responses across forecasting workers

code/probe_tavily_cache.py

@@ -0,0 +1,113 @@
+#!/usr/bin/env python3
+"""Opt-in live cache probe: concurrent tools, no model calls, <=4 HTTP attempts."""
+import argparse
+from collections import Counter
+from concurrent.futures import ProcessPoolExecutor
+import fcntl
+import hashlib
+import importlib
+import json
+import os
+from pathlib import Path
+import sys
+import tempfile
+from unittest.mock import patch
+import urllib.request
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path[:0] = [str(ROOT), str(ROOT / "packages/fri-forecaster")]
+
+
+def call_tool(task):
+    module, tool, args, log = task
+    tools = importlib.import_module(module)
+    original = urllib.request.urlopen
+
+    def counted(req, *args, **kwargs):
+        # Cross-process accounting and a hard budget, even if upstream fails.
+        with open(log, "a+") as stream:
+            fcntl.flock(stream, fcntl.LOCK_EX)
+            stream.seek(0)
+            if len(stream.readlines()) >= 4:
+                raise TimeoutError("probe HTTP budget exhausted")
+            stream.write(req.full_url + "\n")
+            stream.flush()
+        return original(req, *args, **kwargs)
+
+    with patch("urllib.request.urlopen", side_effect=counted):
+        return getattr(tools, tool)(**args)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--live", action="store_true", help="authorize paid Tavily calls")
+    parser.add_argument("--env-file", type=Path, help="read only TAVILY_API_KEY from this file")
+    parser.add_argument("--workers", type=int, default=8)
+    parser.add_argument("--url", default="https://en.wikipedia.org/wiki/Artificial_intelligence")
+    parser.add_argument("--query", default="International AI Safety Report official report")
+    args = parser.parse_args()
+    if not args.live:
+        parser.error("pass --live to authorize real Tavily requests")
+    if not 2 <= args.workers <= 16:
+        parser.error("--workers must be between 2 and 16")
+    if args.env_file:
+        from fri_forecaster.keys import parse_env
+        key = parse_env(args.env_file).get("TAVILY_API_KEY")
+        if key:
+            os.environ["TAVILY_API_KEY"] = key
+    if not os.environ.get("TAVILY_API_KEY"):
+        parser.error("TAVILY_API_KEY is required")
+
+    with tempfile.TemporaryDirectory(prefix="tavily-cache-probe-") as temp:
+        log = str(Path(temp) / "http-attempts")
+        os.environ.update(FORECASTER_TAVILY_CACHE_DIR=temp,
+                          FORECASTER_TAVILY_CACHE_TTL="3600",
+                          FORECASTER_BLOCK_URL_PATTERNS="")
+        modules = ["redlines.tools", "fri_forecaster.tools"]
+        batches = []
+        failures = []
+        with ProcessPoolExecutor(max_workers=args.workers) as pool:
+            def batch(tool, arguments):
+                tasks = [(modules[i % 2], tool, kw, log) for i, kw in enumerate(arguments)]
+                results = list(pool.map(call_tool, tasks))
+                # Cause is per caller, not part of the shared factual response.
+                normalized = [{k: v for k, v in r.items() if k != "cause"} for r in results]
+                same = all(r == normalized[0] for r in normalized)
+                errors = [r.get("error") for r in results if r.get("error")]
+                if errors or not same:
+                    failures.append({"tool": tool, "errors": errors, "identical": same})
+                batches.append({"tool": tool, "calls": len(results), "identical": same,
+                                "successful": len(results) - len(errors),
+                                "response_sha256": hashlib.sha256(json.dumps(
+                                    normalized[0], sort_keys=True).encode()).hexdigest()})
+                return results
+
+            pages = batch("read_page", [{"url": args.url}] * args.workers)
+            if not failures and pages[0].get("next_offset"):
+                batch("read_page", [{"url": args.url, "offset": pages[0]["next_offset"]}]
+                      * args.workers)
+            else:
+                failures.append({"pagination": "page failed or too short to test pagination"})
+            if not failures:
+                searches = batch("web_search", [dict(query=args.query, cause=f"group-{i}")
+                                                for i in range(args.workers)])
+                if any(r.get("cause") != f"group-{i}" for i, r in enumerate(searches)):
+                    failures.append({"cause": "caller tags were not preserved"})
+                if not searches[0].get("results"):
+                    failures.append({"search": "no source results returned"})
+        attempts = Counter(Path(log).read_text().splitlines()) if Path(log).exists() else Counter()
+        expected = {"https://api.tavily.com/extract": 1, "https://api.tavily.com/search": 1}
+        if dict(attempts) != expected:
+            failures.append({"expected_http_attempts": expected})
+        total = sum(b["calls"] for b in batches)
+        report = {"passed": not failures, "tool_calls": total,
+                  "http_attempts": dict(attempts),
+                  "avoided_http_attempts": total - sum(attempts.values()),
+                  "batches": batches, "failures": failures,
+                  "model_calls": 0, "cache": "isolated temporary directory, removed after probe"}
+        print(json.dumps(report, indent=2))
+        return int(bool(failures))
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

packages/fri-forecaster/README.md

@@ -175,6 +175,43 @@ git clone https://github.com/forecastingresearch/agent-box && cd agent-box
 npm ci && python -m venv .venv && .venv/bin/pip install -e '.[dev]'
 ```
 
+## Tavily response cache
+
+Searches and full-page extracts are cached on disk for **one hour** by default,
+shared by all local workers and tool subprocesses (including separate conditions
+and models). Identical concurrent requests make only one upstream call. This is
+implemented in the Tavily tools, not the agent-box service; AIRO's mirrored tools
+use the same cache format and directory.
+
+- `FORECASTER_TAVILY_CACHE_DIR`: defaults to
+  `${XDG_CACHE_HOME:-~/.cache}/fri-forecaster/tavily`. Use a local filesystem.
+- `FORECASTER_TAVILY_CACHE_TTL`: lifetime in seconds; default `3600`, `0` disables.
+  To isolate a forecasting run, set a fresh directory for that run and remove it
+  afterwards. Otherwise nearby runs may reuse evidence for up to one hour.
+- Keys include the endpoint, complete request settings (including news dates),
+  and a hash of the API credential; credentials themselves are never stored.
+  Queries are matched exactly, not semantically. Full-page extraction is shared
+  across offsets; condition/group tags do not partition identical research.
+- URL guards and group tags are applied anew on every call, including cache hits.
+  Tool results and evidence recording retain their existing shape.
+- Network failures, API error responses, and empty/failed extracts are not cached.
+  Cache storage failures fall back to live requests. No stale-on-error fallback.
+- The private cache contains source text. Expired rows are pruned on successful
+  writes; remove the directory when no workers are running to purge everything.
+  This stdlib implementation uses Linux/POSIX advisory file locks.
+
+An opt-in live probe from the AIRO checkout checks eight concurrent workers
+across both tool copies, pagination, and per-caller tags, with an isolated cache:
+
+```sh
+python3 code/probe_tavily_cache.py --live --env-file /path/to/provider.env
+```
+
+It makes no model calls and caps HTTP attempts at four (normally one search and
+one extraction). It prints request counts and checksums, not credentials or page
+contents, and removes its temporary cache afterwards. This verifies deliberate
+identical requests, not the hit rate of a natural forecasting run.
+
 ## Use
 
 ```python

packages/fri-forecaster/fri_forecaster/tools.py

@@ -1,8 +1,6 @@
 #!/usr/bin/env python3
 """Grounding tools the forecasting models can call during a tool-use run.
 
-Moved from AIRO's redlines/tools.py on 2026-10-05 (AIRO re-exports it).
-
 VENDORED on 2026-08-28 from xrisk-canaries' `forecast/cruxgen/tools.py` -- see
 redlines/llm.py for why the move happened. Pure stdlib: urllib against Tavily,
 no third-party client.
@@ -44,11 +42,17 @@
 2026-09-02) and needed METACULUS_API_KEY, which redlines.llm no longer loads.
 The paper tells the same story in its methods section.
 """
+import fcntl
+import hashlib
 import json
 import os
 import re
+import sqlite3
+import time
+from pathlib import Path
 import urllib.error
 import urllib.request
+from contextlib import closing
 from datetime import date, timedelta
 
 TAVILY_URL = "https://api.tavily.com/search"
@@ -103,6 +107,71 @@ def search_body(query, max_results=5, recent_days=None, today=None):
     return body
 
 
+def _cached_tavily(endpoint, spec, key, fetch):
+    """Cache successful raw responses across processes; never cache credentials.
+
+    A bounded set of advisory locks coalesces simultaneous misses. Filtering,
+    cause tags and page windows happen AFTER this layer, on every invocation.
+    Cache/storage failures fail open, but fetch exceptions propagate normally.
+    """
+    try:
+        ttl = float(os.environ.get("FORECASTER_TAVILY_CACHE_TTL", "3600"))
+    except ValueError:
+        ttl = 0
+    if not 0 < ttl < float("inf"):
+        return fetch()
+    lock = None
+    try:
+        root = Path(os.environ.get("FORECASTER_TAVILY_CACHE_DIR") or
+                    str(Path(os.environ.get("XDG_CACHE_HOME") or
+                             Path.home() / ".cache") / "fri-forecaster" / "tavily"))
+        root.mkdir(parents=True, exist_ok=True, mode=0o700)
+        # Hash the account into the key to avoid sharing results across keys;
+        # neither the key nor request credentials are written to disk.
+        digest = hashlib.sha256(json.dumps(
+            ["v1", endpoint, spec, hashlib.sha256(key.encode()).hexdigest()],
+            sort_keys=True).encode()).hexdigest()
+        lock = (root / ("lock-" + str(int(digest[:2], 16) % 64))).open("a")
+        fcntl.flock(lock, fcntl.LOCK_EX)
+    except OSError:
+        if lock is not None:
+            lock.close()
+        return fetch()
+    try:
+        now = time.time()
+        try:
+            with closing(sqlite3.connect(root / "responses.sqlite3", timeout=5)) as db, db:
+                db.execute("CREATE TABLE IF NOT EXISTS responses "
+                           "(key TEXT PRIMARY KEY, created REAL, payload TEXT)")
+                row = db.execute("SELECT created, payload FROM responses WHERE key=?",
+                                 (digest,)).fetchone()
+            if row and 0 <= now - row[0] < ttl:
+                return json.loads(row[1])
+        except (OSError, sqlite3.Error, ValueError):
+            pass
+        data = fetch()
+        # Empty/failed extracts and API error envelopes must remain retryable.
+        if (isinstance(data, dict) and not data.get("error") and
+                isinstance(data.get("results"), list) and
+                (endpoint != TAVILY_EXTRACT_URL or
+                 any(r.get("raw_content") for r in data["results"]))):
+            try:
+                with closing(sqlite3.connect(root / "responses.sqlite3", timeout=5)) as db, db:
+                    db.execute("DELETE FROM responses WHERE created < ?", (now - ttl,))
+                    db.execute("INSERT OR REPLACE INTO responses VALUES (?, ?, ?)",
+                               (digest, now, json.dumps(data)))
+            except (OSError, sqlite3.Error, ValueError):
+                pass
+        return data
+    finally:
+        lock.close()
+
+
+def _fetch_tavily(req, timeout):
+    with urllib.request.urlopen(req, timeout=timeout) as resp:
+        return json.load(resp)
+
+
 def web_search(query, max_results=5, recent_days=None, cause=None):
     """Search the web via Tavily. Returns {query, answer?, results:[{title,url,
     content, published?}]}. `recent_days` bounds the search to news published in
@@ -127,8 +196,8 @@ def web_search(query, max_results=5, recent_days=None, cause=None):
     req = urllib.request.Request(
         TAVILY_URL, data=body, headers={"Content-Type": "application/json"})
     try:
-        with urllib.request.urlopen(req, timeout=30) as resp:
-            data = json.load(resp)
+        data = _cached_tavily(TAVILY_URL, spec, key,
+                              lambda: _fetch_tavily(req, 30))
     except (urllib.error.URLError, ValueError, TimeoutError) as e:
         return {"query": query, **tag, "results": [], "error": f"web search failed: {e}"}
     out = {"query": query, **tag, "answer": data.get("answer"), "results": []}
@@ -164,13 +233,14 @@ def read_page(url, offset=0):
     if not key:
         return {"url": url, "text": "",
                 "error": "page reading unavailable (no TAVILY_API_KEY set)"}
-    body = json.dumps({"urls": [url], "extract_depth": "basic"}).encode()
+    spec = {"urls": [url], "extract_depth": "basic"}
+    body = json.dumps(spec).encode()
     req = urllib.request.Request(
         TAVILY_EXTRACT_URL, data=body,
         headers={"Content-Type": "application/json", "Authorization": f"Bearer {key}"})
     try:
-        with urllib.request.urlopen(req, timeout=45) as resp:
-            data = json.load(resp)
+        data = _cached_tavily(TAVILY_EXTRACT_URL, spec, key,
+                              lambda: _fetch_tavily(req, 45))
     except (urllib.error.URLError, ValueError, TimeoutError) as e:
         return {"url": url, "text": "", "error": f"page read failed: {e}"}
     results = data.get("results") or []

packages/fri-forecaster/tests/test_url_guard.py

@@ -28,7 +28,7 @@ def reply(data):
 class UrlGuardTest(unittest.TestCase):
     def env(self, patterns=None):
         """TAVILY_API_KEY set; the guard variable unset (None) or set to `patterns`."""
-        env = {"TAVILY_API_KEY": "k"}
+        env = {"TAVILY_API_KEY": "k", "FORECASTER_TAVILY_CACHE_TTL": "0"}
         if patterns is not None:
             env[tools.BLOCK_ENV] = patterns
         return patch.dict(os.environ, env, clear=True)