diff --git a/README.md b/README.md index 4b171e7..d641b0c 100644 --- a/README.md +++ b/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 diff --git a/fri_forecaster/tools.py b/fri_forecaster/tools.py index b8fa8a1..a1300ec 100644 --- a/fri_forecaster/tools.py +++ b/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": "", diff --git a/tests/test_pi_tools.py b/tests/test_pi_tools.py index f83c4a9..101f15f 100644 --- a/tests/test_pi_tools.py +++ b/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() diff --git a/tests/test_url_guard.py b/tests/test_url_guard.py new file mode 100644 index 0000000..7727144 --- /dev/null +++ b/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()