Coverage for src/local_deep_research/advanced_search_system/tools/archive.py: 98%
86 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""Fetch a historical URL through the existing page-fetch pipeline."""
3from typing import Any
5from langchain_core.tools import tool
6from loguru import logger
8from ...security import redact_url_for_log, sanitize_error_for_agent
9from ...security.egress.policy import PolicyDeniedError, evaluate_engine
10from ...utilities.type_utils import resolve_boolean_setting
11from ...web_search_engines.engine_availability import failure_from_exception
12from ...web_search_engines.wayback_client import (
13 AVAILABLE_API,
14 lookup_archive_snapshot,
15 normalize_archive_date,
16 validate_capture_fetch_url,
17 validate_archive_url,
18)
19from .fetch import (
20 FETCH_MODES,
21 _check_and_record_fetch,
22 _enforce_url_policy,
23 _read_content_max_length_setting,
24 _read_max_fetches_setting,
25 _read_summary_input_max_length_setting,
26 _summary_invocation_budget,
27 build_fetch_tool,
28)
31def _enforce_archive_engine(egress_context, settings_snapshot):
32 if egress_context is not None:
33 decision = evaluate_engine(
34 "wayback", egress_context, settings_snapshot=settings_snapshot or {}
35 )
36 if not decision.allowed:
37 raise PolicyDeniedError(decision, target=AVAILABLE_API)
40def build_archive_fetch_tool(
41 mode: str,
42 collector: Any,
43 *,
44 model=None,
45 overall_query: str = "",
46 settings_snapshot: dict | None = None,
47 egress_context=None,
48 max_fetches: int | None = None,
49 subagent_id: str | None = None,
50 availability=None,
51):
52 """Build fetch_archive, following the run's fetch mode and shared quota."""
53 if mode not in FETCH_MODES:
54 raise ValueError(f"Unknown fetch mode {mode!r}")
55 if mode == "disabled":
56 return None
57 if not resolve_boolean_setting(
58 settings_snapshot, "search.engine.web.wayback.agent_enabled", True
59 ):
60 return None
61 if (
62 egress_context is not None
63 and not evaluate_engine(
64 "wayback", egress_context, settings_snapshot=settings_snapshot or {}
65 ).allowed
66 ):
67 return None
68 if mode != "full" and model is None:
69 raise ValueError(f"{mode} fetch mode requires a model")
70 limit = (
71 max_fetches
72 if max_fetches is not None
73 else _read_max_fetches_setting(settings_snapshot)
74 )
75 description = (
76 "Read an archived version of one absolute HTTP(S) URL. Discover URLs "
77 "with a configured search provider first, or use a URL supplied by the user. "
78 "Optionally pass date as YYYY-MM-DD to select the closest available capture; "
79 "omit date for the most recent available capture. The actual capture date "
80 "can differ from the requested date. Never substitutes the live page. "
81 "Shares the fetch_content budget and fetch mode."
82 )
83 if mode != "full":
84 description += " Pass focus as the specific question to extract from the archived page."
86 def fetch(url: str, date: str | None, focus: str | None = None) -> str:
87 try:
88 url = validate_archive_url(url)
89 timestamp = normalize_archive_date(date)
90 if availability is not None:
91 state = availability.state("wayback")
92 if state is not None:
93 return state.describe()
94 _enforce_archive_engine(egress_context, settings_snapshot)
95 # Check the target before sending it to an external archive, and
96 # gate the API before reserving the shared quota slot.
97 _enforce_url_policy(url, egress_context)
98 _enforce_url_policy(AVAILABLE_API, egress_context)
99 # Refuse an invalid configured budget before archive lookup or
100 # reservation of a shared fetch slot. The delegated page tool
101 # applies the same budget to the selected capture.
102 if mode == "full":
103 _read_content_max_length_setting(settings_snapshot)
104 else:
105 _read_summary_input_max_length_setting(settings_snapshot)
106 _summary_invocation_budget(model, settings_snapshot)
107 error, count = _check_and_record_fetch(
108 collector, limit, subagent_id=subagent_id
109 )
110 if error is not None:
111 return error
112 snapshot = lookup_archive_snapshot(url, timestamp)
113 if snapshot is None:
114 if availability is not None: 114 ↛ 115line 114 didn't jump to line 115 because the condition on line 114 was never true
115 availability.record_success("wayback")
116 return f"No accessible archive capture was found for {url}. Try another date or URL."
117 replay = snapshot["url"]
118 embedded_target = replay.split(f"/web/{snapshot['timestamp']}/", 1)[
119 1
120 ]
121 _enforce_url_policy(embedded_target, egress_context)
122 _enforce_url_policy(replay, egress_context)
123 metadata = {
124 "source_type": "archive",
125 "original_url": snapshot["original_url"],
126 "requested_url": snapshot["requested_url"],
127 "timestamp": snapshot["timestamp"],
128 "formatted_date": snapshot["formatted_date"],
129 }
131 def capture_guard(candidate):
132 validate_capture_fetch_url(
133 candidate, snapshot["timestamp"], snapshot["original_url"]
134 )
136 page_fetch = build_fetch_tool(
137 mode,
138 collector,
139 model=model,
140 overall_query=overall_query,
141 settings_snapshot=settings_snapshot,
142 egress_context=egress_context,
143 max_fetches=limit,
144 subagent_id=subagent_id,
145 record_fetch=False,
146 source_metadata=metadata,
147 url_guard=capture_guard,
148 availability=availability,
149 )
150 arguments: dict[str, Any] = {"url": replay}
151 if mode != "full":
152 arguments["focus"] = focus
153 message = page_fetch.invoke(arguments)
154 # Shared fetching displays provenance on successful reads,
155 # including ordinary rereads of the resulting citation.
156 if date is not None:
157 message += (
158 f"\nRequested date: {date} (closest available capture)"
159 )
160 if limit > 0 and count is not None:
161 message += f"\n\n({count} of {limit} page fetches used)"
162 return message
163 except PolicyDeniedError as exc:
164 return sanitize_error_for_agent(
165 f"Cannot fetch archive: blocked by egress policy ({exc.decision.reason}). "
166 "Use an allowed source or choose an appropriate search provider."
167 )
168 except Exception as exc:
169 logger.opt(exception=False).warning(
170 "Archive fetch failed for {} ({})",
171 redact_url_for_log(url),
172 type(exc).__name__,
173 )
174 failure = failure_from_exception(exc)
175 if availability is not None and failure is not None:
176 return availability.disable("wayback", failure).describe()
177 return sanitize_error_for_agent(f"Cannot fetch archive: {exc}")
179 if mode == "full":
181 @tool("fetch_archive", description=description)
182 def fetch_archive(url: str, date: str | None = None) -> str:
183 """Read the archived page at URL, optionally near a calendar date."""
184 return fetch(url, date)
186 return fetch_archive
188 @tool("fetch_archive", description=description)
189 def fetch_archive_summary(
190 url: str, focus: str, date: str | None = None
191 ) -> str:
192 """Extract facts relevant to focus from an archived page."""
193 return fetch(url, date, focus)
195 return fetch_archive_summary