[eric] web: fall back to the Wayback Machine when the live page is dead or walled

backend/apps/agents/tools/fetch/wayback.py

@@ -0,0 +1,55 @@
+"""Wayback Machine fallback for URLs the live web won't hand over.
+
+When a page is gone (404, domain dead, article pulled) or sits behind a wall
+that no client-side trick beats, the archive usually still has the REAL text,
+which beats a grounded model's summary of a page it also couldn't read.
+
+Uses `web.archive.org/web/2/<url>`, which redirects to the closest snapshot.
+The documented `archive.org/wayback/available` JSON API is NOT used: it is
+aggressively throttled and answered 429 on every probe from this machine, while
+the redirect path answered in 0.5-3s.
+"""
+
+import re
+from typing import Optional
+from urllib.parse import urlparse
+
+from typeguard import typechecked
+
+from backend.apps.agents.tools.browser_http import browser_request
+from backend.apps.agents.tools.fetch.page_text import html_to_text
+
+P_WAYBACK_LATEST = "https://web.archive.org/web/2/"
+P_ALLOWED_HOST = "web.archive.org"
+P_TIMEOUT = 10.0
+# Below this the "snapshot" is a stub or an archived error page, not the article.
+P_MIN_SUBSTANCE_CHARS = 200
+P_SNAPSHOT_RE = re.compile(r"/web/(\d{4})(\d{2})(\d{2})\d*/")
+
+
+@typechecked
+def snapshot_date(archived_url: str) -> Optional[str]:
+    """The snapshot's date, so the model knows how stale the text is."""
+    match = P_SNAPSHOT_RE.search(archived_url)
+    if not match:
+        return None
+    return f"{match.group(1)}-{match.group(2)}-{match.group(3)}"
+
+
+@typechecked
+async def fetch_wayback(url: str) -> Optional[str]:
+    """The archived page text, or None when there is no usable snapshot."""
+    reply = await browser_request(P_WAYBACK_LATEST + url, timeout=P_TIMEOUT)
+    # We hand the archive a caller-supplied URL, so confirm we actually ended up on the archive and not somewhere it redirected us.
+    if urlparse(reply.url).hostname != P_ALLOWED_HOST:
+        return None
+    if reply.status != 200:
+        return None
+    text = html_to_text(reply.text).strip()
+    if len(text) < P_MIN_SUBSTANCE_CHARS:
+        return None
+    taken = snapshot_date(reply.url)
+    header = f"Archived copy of {url}"
+    header += f" (Wayback Machine snapshot from {taken}); the live page could not be read." if taken \
+        else " (Wayback Machine); the live page could not be read."
+    return f"{header}\n\n{text}"

backend/apps/web/web.py

@@ -69,6 +69,7 @@ KEYLESS_TIER_SECONDS = 8.0        # a search frontend answers in ~1s; >8s is a h
 BROWSER_TIER_SECONDS = 12.0       # the main-bridge send has its own per-action timeout
 GROUNDED_TIER_SECONDS = 45.0      # grounded native search legitimately takes 30-42s
 LOCAL_FETCH_TIER_SECONDS = 15.0   # normal pages return in <2s
+ARCHIVE_TIER_SECONDS = 10.0       # the Wayback redirect path answered in 0.5-3s
 
 
 # --------------------------------------------------------------------------- Helpers ---------------------------------------------------------------------------
@@ -95,7 +96,7 @@ async def p_browser_bridge(action: str, params: Dict) -> Optional[Dict]:
     return res
 
 
-# When every search backend fails, point the model at the in-product browser (always-on CreateBrowserAgent tool) instead of telling it to "wait and retry", which it can't do and just relays as a dead end. The real Chromium renders pages and isn't subject to the scrape throttle.
+# When every search backend fails, point the model at the in-product browser (always-on CreateBrowserAgent tool) instead of telling it to "wait and retry", which it can't do and just relays as a dead end. A real Chromium carries a real browser fingerprint, which is what the challenge is actually keyed on.
 @typechecked
 def p_browser_fallback_nudge(query: str) -> str:
     return (
@@ -272,6 +273,14 @@ async def fetch(body: FetchBody) -> Dict:
             return None
         return {"url": body.url, "content": f"Contents of {body.url}:\n\n{res['text']}", "backend": "browser"}
 
+    async def try_wayback() -> Optional[Dict]:
+        # A dead link or a hard bot wall is exactly what the archive is for, and it returns the page's REAL text where a grounded fetcher can only summarise a page it also can't read.
+        from backend.apps.agents.tools.fetch.wayback import fetch_wayback
+        text = await fetch_wayback(body.url)
+        if not text:
+            return None
+        return {"url": body.url, "content": text, "backend": "wayback"}
+
     async def try_gemini() -> Optional[Dict]:
         if not gemini_key:
             return None
@@ -313,6 +322,7 @@ async def fetch(body: FetchBody) -> Dict:
     tiers = [
         CascadeTier(name="local", run=try_local, budget=LOCAL_FETCH_TIER_SECONDS),
         CascadeTier(name="browser", run=try_browser_fetch, budget=BROWSER_TIER_SECONDS),
+        CascadeTier(name="wayback", run=try_wayback, budget=ARCHIVE_TIER_SECONDS),
     ] + p_grounded_tiers("fetch", body.primary, {
         "gemini_native": try_gemini,
         "gemini_subscription": try_gemini_subscription,

backend/tests/test_web_cascade_budget.py

@@ -127,5 +127,5 @@ def test_search_tier_budgets_leave_room_for_the_grounded_tier():
     grounded_floor = 20.0
     search_cheap = 2 * W.KEYLESS_TIER_SECONDS + W.BROWSER_TIER_SECONDS
     assert W.SEARCH_BUDGET_SECONDS - search_cheap >= grounded_floor
-    fetch_cheap = W.LOCAL_FETCH_TIER_SECONDS + W.BROWSER_TIER_SECONDS
+    fetch_cheap = W.LOCAL_FETCH_TIER_SECONDS + W.BROWSER_TIER_SECONDS + W.ARCHIVE_TIER_SECONDS
     assert W.FETCH_BUDGET_SECONDS - fetch_cheap >= grounded_floor

backend/tests/test_web_search_cascade.py

@@ -15,6 +15,7 @@ import time
 
 import pytest
 
+import backend.apps.agents.tools.fetch.wayback as WB
 import backend.apps.agents.tools.search.search_startpage as SP
 import backend.apps.web.web as W
 from backend.apps.web.web import search, SearchBody
@@ -41,6 +42,10 @@ def p_no_network(monkeypatch):
         return None
     monkeypatch.setattr(SP, "search_startpage", p_startpage_closed)
 
+    async def p_no_snapshot(url):
+        return None
+    monkeypatch.setattr(WB, "fetch_wayback", p_no_snapshot)
+
 
 def p_ddg_returns(monkeypatch, text):
     async def p_f(query, num):
@@ -244,6 +249,24 @@ async def test_fetch_local_error_returned_as_last_resort(monkeypatch):
     assert "HTTP error 403" in res["content"]
 
 
+@pytest.mark.asyncio
+async def test_fetch_falls_to_the_archive_when_the_live_page_is_gone(monkeypatch):
+    """A dead link is exactly what the archive is for; the real text beats a grounded summary of nothing."""
+    p_local_returns(monkeypatch, "HTTP error 404 fetching https://gone.example/post")
+
+    async def p_snapshot(url):
+        return "Archived copy of https://gone.example/post (Wayback Machine snapshot from 2026-05-08)\n\nThe original text."
+    monkeypatch.setattr(WB, "fetch_wayback", p_snapshot)
+
+    async def p_boom(*a, **k):
+        raise AssertionError("a paid fetcher must not run once the archive answered")
+    monkeypatch.setattr(W, "gemini_grounded_call", p_boom)
+
+    res = await fetch(FetchBody(url="https://gone.example/post"))
+    assert res["backend"] == "wayback"
+    assert "The original text." in res["content"]
+
+
 # --- packaged-browser tier: fires when DDG throttles, skipped when no bridge ---
 
 def p_browser_bridge(monkeypatch, result):