URL guard: FORECASTER_BLOCK_URL_PATTERNS, off unless set

README.md

@@ -161,6 +161,10 @@ dropped.
   or its `cli.js`). pi owns model auth.
 - `TAVILY_API_KEY` for research, and the providers' API keys, from the environment or an
   env file (`keys.py`).
+- Optional: `FORECASTER_BLOCK_URL_PATTERNS`, regular expressions separated by commas or
+  newlines, matched case-insensitively against URLs. When set, `web_search` drops matching
+  results and `read_page` refuses matching URLs; unset or empty, nothing changes. Pass it
+  per run with `harness.run(env=...)`.
 
 ```sh
 git clone https://github.com/forecastingresearch/agent-box && cd agent-box

fri_forecaster/tools.py

@@ -46,6 +46,7 @@
 """
 import json
 import os
+import re
 import urllib.error
 import urllib.request
 from datetime import date, timedelta
@@ -58,6 +59,29 @@
 # its envelope; the model pages through a longer document with `offset`.
 PAGE_CHARS = 7000
 
+# URL GUARD (2026-10-16, AIRO's anchor-guard experiment). OFF unless the
+# environment sets FORECASTER_BLOCK_URL_PATTERNS: regexes separated by newlines
+# or commas (so no commas inside a pattern), searched case-insensitively
+# against each URL. pi_tools.ts hands its
+# environment to every tool process, so one variable reaches both tools.
+# read_page refuses a matching URL without calling Tavily, with the ordinary
+# failure shape and a neutral reason; web_search drops matching results, and
+# results titled as the AIRO paper (BLOCK_TITLES), and blanks Tavily's
+# `answer` when it dropped anything, because that summary was written from
+# the unfiltered results. ForecastBench shares these tools, so nothing here
+# changes when the variable is unset.
+BLOCK_ENV = "FORECASTER_BLOCK_URL_PATTERNS"
+BLOCK_TITLES = ("Automated Artificial Intelligence Risk Outlook",)
+
+
+def blocked_patterns():
+    raw = os.environ.get(BLOCK_ENV) or ""
+    return [re.compile(p.strip(), re.I) for p in re.split(r"[\n,]", raw) if p.strip()]
+
+
+def is_blocked(url, patterns):
+    return bool(url) and any(p.search(url) for p in patterns)
+
 
 def search_body(query, max_results=5, recent_days=None, today=None):
     """The Tavily request for a search, minus the key. With `recent_days` the
@@ -108,9 +132,14 @@ def web_search(query, max_results=5, recent_days=None, cause=None):
     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": []}
+    guard = blocked_patterns()
     if recent_days:
         out["window"] = {"start": spec["start_date"], "end": spec["end_date"]}
     for r in data.get("results", []):
+        if guard and (is_blocked(r.get("url"), guard)
+                      or any(t in (r.get("title") or "") for t in BLOCK_TITLES)):
+            out["answer"] = None
+            continue
         item = {"title": r.get("title"), "url": r.get("url"),
                 # Tavily already returns trimmed snippets; cap defensively
                 "content": (r.get("content") or "")[:1500]}
@@ -129,6 +158,8 @@ def read_page(url, offset=0):
     page cannot be fetched, returns a result carrying an `error` string and no
     text, so the loop keeps going.
     """
+    if is_blocked(url, blocked_patterns()):
+        return {"url": url, "text": "", "error": "page read failed: page unavailable"}
     key = os.environ.get("TAVILY_API_KEY")
     if not key:
         return {"url": url, "text": "",

tests/test_pi_tools.py

@@ -135,6 +135,7 @@ def run_pi(self, script, **kw):
                     cell_schema=SCHEMA, key_fields=["k"], allowed_keys=["a", "b"], tool_descriptions=DESC,
                     min_research=2, thinking="off", timeout=120,
                     env={"PI_CODING_AGENT_DIR": agent, "PI_OFFLINE": "1", "PI_TELEMETRY": "0"})
+        args["env"].update(kw.pop("env", {}))
         args.update(kw)
         return harness.run(**args), model
 
@@ -290,5 +291,26 @@ def test_debrief_asks_again_after_a_refusal(self):
                                           "attempts": 2, "problems": [], "error": None})
 
 
+@NEEDS_PI
+class UrlGuardPassthroughTest(PiCase):
+    """FORECASTER_BLOCK_URL_PATTERNS handed to harness.run(env=...) reaches the
+    tool process: pi, pi_tools.ts and the real tool_cli, down to tools.py."""
+
+    def test_env_reaches_read_page(self):
+        harness.TOOL_CLI = self.real_cli    # the real CLI; with no Tavily key nothing is fetched
+        res, model = self.run_pi([
+            ("read_page", {"url": "https://example.com/paper.pdf"}),         # 1: blocked by the guard
+            ("read_page", {"url": "https://example.org/other"}),             # 2: passes it, stops at the key
+            ("submit_forecast", {"forecasts": [{"k": "a", "probability": 0.5}, {"k": "b", "probability": 0.5}],
+                                 "rationale": "r"}),
+            "done",
+        ], min_research=0, env={"FORECASTER_BLOCK_URL_PATTERNS": "example\\.com",
+                                "TAVILY_API_KEY": "", "FORECASTER_ENV_FILE": os.path.join(self.tmp.name, "no-env")})
+        self.assertEqual(res["rc"], 0, open(os.path.join(res["workdir"], "stderr.txt")).read())
+        self.assertIn("page read failed: page unavailable", model.tool_result(2))
+        self.assertIn("no TAVILY_API_KEY set", model.tool_result(3))
+        self.assertEqual((res["usage"]["research_ok"], res["usage"]["research_failed"]), (0, 2))
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_url_guard.py

@@ -0,0 +1,111 @@
+"""fri_forecaster.tools URL guard (FORECASTER_BLOCK_URL_PATTERNS), with the
+network mocked: off unless set; read_page refuses without calling Tavily;
+web_search drops matching results and the answer built from them."""
+import io
+import json
+import os
+import unittest
+from unittest.mock import patch
+
+from fri_forecaster import tools
+
+PATTERNS = ("forecastingresearch\\.org/pdf/airo\n"
+            "forecastingresearch\\.org/research/airo,airo\\.forecastingresearch\\.org")
+PAPER = "https://forecastingresearch.org/pdf/airo-working-paper.pdf"
+SEARCH = {"answer": "AIRO says 0.47%", "results": [
+    {"title": "AIRO working paper", "url": PAPER, "content": "x"},
+    {"title": "Automated Artificial Intelligence Risk Outlook (mirror)",
+     "url": "https://example.com/mirror", "content": "x"},
+    {"title": "XPT", "url": "https://forecastingresearch.org/xpt", "content": "x"},
+    {"title": "Launch post", "url": "https://www.lesswrong.com/posts/abc/airo", "content": "x"},
+]}
+
+
+def reply(data):
+    return io.BytesIO(json.dumps(data).encode())
+
+
+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"}
+        if patterns is not None:
+            env[tools.BLOCK_ENV] = patterns
+        return patch.dict(os.environ, env, clear=True)
+
+    def test_patterns_unset_empty_one_several(self):
+        for raw in (None, "", " \n , "):
+            with self.env(raw):
+                self.assertEqual(tools.blocked_patterns(), [], repr(raw))
+        with self.env("airo\\.forecastingresearch\\.org"):
+            one = tools.blocked_patterns()
+        self.assertEqual(len(one), 1)
+        self.assertTrue(tools.is_blocked("https://AIRO.ForecastingResearch.org/x", one))
+        self.assertFalse(tools.is_blocked("https://forecastingresearch.org/xpt", one))
+        self.assertFalse(tools.is_blocked(None, one))
+        with self.env(PATTERNS):                       # newline- and comma-separated
+            self.assertEqual(len(tools.blocked_patterns()), 3)
+        self.assertFalse(tools.is_blocked(PAPER, []))  # no patterns: nothing blocked
+
+    def test_one_pattern_blocks_only_its_matches(self):
+        page = {"results": [{"raw_content": "XPT text"}]}
+        one = "forecastingresearch\\.org/pdf/airo"
+        with self.env(one), patch("urllib.request.urlopen", return_value=reply(page)) as net:
+            self.assertEqual(tools.read_page(PAPER)["error"], "page read failed: page unavailable")
+            self.assertEqual(tools.read_page(PAPER.upper())["error"], "page read failed: page unavailable")
+            self.assertEqual(tools.read_page("https://forecastingresearch.org/research/airo")["text"], "XPT text")
+        net.assert_called_once()
+        search = {"answer": "from the paper", "results": [
+            {"title": "AIRO working paper", "url": PAPER, "content": "x"},
+            {"title": "Mirror", "url": PAPER.upper(), "content": "x"},
+            {"title": "XPT", "url": "https://forecastingresearch.org/xpt", "content": "x"},
+            {"title": "Research index", "url": "https://forecastingresearch.org/research/airo", "content": "x"},
+        ]}
+        with self.env(one), patch("urllib.request.urlopen", return_value=reply(search)):
+            out = tools.web_search("ai risk")
+        self.assertEqual([r["url"] for r in out["results"]],
+                         ["https://forecastingresearch.org/xpt", "https://forecastingresearch.org/research/airo"])
+        self.assertIsNone(out["answer"])
+
+    def test_read_page_blocked_without_calling_tavily(self):
+        with self.env(PATTERNS), patch("urllib.request.urlopen") as net:
+            for url in (PAPER, "https://forecastingresearch.org/research/airo",
+                        "https://airo.forecastingresearch.org/",
+                        "https://Airo.ForecastingResearch.org/", PAPER.replace("airo", "AIRO")):
+                self.assertEqual(tools.read_page(url), {
+                    "url": url, "text": "", "error": "page read failed: page unavailable"})
+        net.assert_not_called()
+
+    def test_read_page_others_still_fetched(self):
+        page = {"results": [{"raw_content": "XPT text"}]}
+        for env in (self.env(PATTERNS), self.env()):
+            with env, patch("urllib.request.urlopen", return_value=reply(page)) as net:
+                self.assertEqual(tools.read_page("https://forecastingresearch.org/xpt")["text"], "XPT text")
+            net.assert_called_once()
+        with self.env(), patch("urllib.request.urlopen", return_value=reply(page)):
+            self.assertEqual(tools.read_page(PAPER)["text"], "XPT text")   # off when unset
+
+    def test_web_search_drops_matches_and_answer(self):
+        with self.env(PATTERNS), patch("urllib.request.urlopen", return_value=reply(SEARCH)):
+            out = tools.web_search("ai risk")
+        self.assertEqual([r["url"] for r in out["results"]],
+                         ["https://forecastingresearch.org/xpt", "https://www.lesswrong.com/posts/abc/airo"])
+        self.assertIsNone(out["answer"])
+
+    def test_web_search_keeps_answer_when_nothing_dropped(self):
+        clean = {"answer": "ok", "results": SEARCH["results"][2:]}
+        with self.env(PATTERNS), patch("urllib.request.urlopen", return_value=reply(clean)):
+            out = tools.web_search("ai risk")
+        self.assertEqual(out["answer"], "ok")
+        self.assertEqual(len(out["results"]), 2)
+
+    def test_web_search_unchanged_when_unset_or_empty(self):
+        for env in (self.env(), self.env("")):
+            with env, patch("urllib.request.urlopen", return_value=reply(SEARCH)):
+                out = tools.web_search("ai risk")
+            self.assertEqual(out["answer"], "AIRO says 0.47%")
+            self.assertEqual(len(out["results"]), 4)
+
+
+if __name__ == "__main__":
+    unittest.main()