Coverage for src/local_deep_research/web_search_engines/engines/full_search.py: 97%
236 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1import time
2from datetime import datetime, UTC
3from typing import (
4 Any,
5 Callable,
6 Dict,
7 List,
8 Optional,
9 Protocol,
10 runtime_checkable,
11)
13from langchain_core.language_models import BaseLLM
15from ...config.search_config import QUALITY_CHECK_DDG_URLS
16from ..engine_availability import SearchFailure, engine_failure
17from ...research_library.downloaders.extraction import (
18 batch_fetch_and_extract,
19)
20from ...security.egress.policy import EgressScope, PolicyDeniedError
21from ...security.log_sanitizer import scrub_error
22from ...security.secure_logging import logger
23from ...security.ssrf_validator import redact_url_for_log, validate_url
24from ...utilities.js_rendering import (
25 read_js_rendering_setting as _read_js_rendering_setting,
26)
27from ...utilities.json_utils import extract_json, get_llm_response_text
28from ...utilities.type_utils import resolve_boolean_setting
29from ...utilities.llm_utils import invoke_llm_sync
32@runtime_checkable
33class _Invokable(Protocol):
34 def invoke(self, query: str) -> Any: ... 34 ↛ exitline 34 didn't return from function 'invoke' because
37# Minimum seconds between bulk-fetch heartbeat milestones. Each UI
38# emit is a DB row + socket event, so per-page emits would spam the
39# research log on a 46-page fetch; the start announcement is always
40# emitted and the completion milestone ("From the web …") follows the
41# fetch, so heartbeats only fill the silence in between.
42_FETCH_HEARTBEAT_SECONDS = 15.0
45def make_fetch_progress_adapter(progress_callback):
46 """Adapt bulk-fetch ``(done, total)`` progress to UI milestones.
48 Shared by the pipeline-strategy wiring (``AdvancedSearchSystem`` →
49 ``FullSearchResults``) and the langgraph-agent tool factories, so
50 both surfaces throttle identically.
52 Returns a hook for engine page-fetch progress that:
53 * announces the fetch upfront (``done == 0``) so a minutes-long
54 fetch never looks stalled,
55 * emits a heartbeat at most every ``_FETCH_HEARTBEAT_SECONDS``,
56 * stays silent on the final page — the completion milestone
57 ("From the web …") follows immediately,
58 * performs a silent termination check on every page (no UI
59 emission): each check doubles as a cancellation check, because
60 the UI callback raises ``ResearchTerminatedException`` when the
61 user cancels — and that derives ``BaseException``, so it
62 propagates through the fetch pipeline's ``except Exception``
63 handlers untouched. Cancellation during a long fetch therefore
64 takes effect promptly instead of only after the whole batch.
66 ``progress_callback`` is the 3-arg UI callback
67 ``(message, percent, metadata)``; ``None`` disables everything.
68 """
69 state = {"last_emit": 0.0}
71 def _on_fetch_progress(done: int, total: int) -> None:
72 if progress_callback is None or total <= 0:
73 return
74 # Silent cancellation check on every page — no UI/logging.
75 progress_callback(
76 "Checking termination status",
77 None,
78 {"phase": "termination_check"},
79 )
80 if done >= total:
81 return
82 now = time.monotonic()
83 if done > 0 and now - state["last_emit"] < _FETCH_HEARTBEAT_SECONDS:
84 return
85 state["last_emit"] = now
86 if done <= 0:
87 progress_callback(
88 f"📄 Fetching full content for {total} pages — "
89 "this can take a minute…",
90 None,
91 {"phase": "fetch", "type": "milestone", "total": total},
92 )
93 else:
94 progress_callback(
95 f"📄 Fetching page content ({done}/{total})…",
96 None,
97 {
98 "phase": "fetch",
99 "type": "milestone",
100 "done": done,
101 "total": total,
102 },
103 )
105 return _on_fetch_progress
108# Egress scopes whose ``evaluate_url`` admits every http(s) host the SSRF
109# validator admits (UNPROTECTED lifts only the scope restriction; the hard
110# SSRF and cloud-metadata blocks stay and the download pipeline enforces
111# those itself on every destination). Under any other scope the run policy
112# restricts destinations by address class (PUBLIC_ONLY: no private hosts;
113# PRIVATE_ONLY / STRICT: no public hosts; BOTH: classified hosts only),
114# so the operator's private-result grant is withheld there from both result
115# URL validation and the pipeline. The pipeline now checks each redirect and
116# pinned address against the run policy; browser fallback is disabled for
117# restricted scopes because Chromium resolves its own DNS. ADAPTIVE never
118# reaches this module (resolved at context construction); an unknown scope
119# fails closed.
120_SCOPES_WITHOUT_HOST_CLASS_RESTRICTION = frozenset({EgressScope.UNPROTECTED})
123class FullSearchResults:
124 def __init__(
125 self,
126 llm: Optional[BaseLLM],
127 web_search: _Invokable,
128 output_format: str = "list",
129 language: str = "English",
130 max_results: int = 10,
131 region: str = "wt-wt",
132 time: Optional[str] = "y",
133 safesearch: str | int = "Moderate",
134 settings_snapshot: Optional[Dict] = None,
135 egress_context: Optional[Any] = None,
136 allow_private_ips: bool = False,
137 content_cache: Optional[Any] = None,
138 progress_callback: Optional[Callable[[int, int], None]] = None,
139 ):
140 self.llm = llm
141 self.output_format = output_format
142 self.language = language
143 self.max_results = max_results
144 self.region = region
145 self.time = time
146 self.safesearch = safesearch
147 self.web_search = web_search
148 self.settings_snapshot = settings_snapshot
149 # The parent engine's resolved result-fetch grant, handed over by
150 # the factory when it builds this wrapper
151 # (``BaseSearchEngine.allow_private_result_fetch``; e.g. SearXNG:
152 # instance approved for private egress AND the env-only
153 # ``search.allow_private_result_fetch`` opt-in). When True, SSRF
154 # validation of result URLs may pass private/loopback IPs; the
155 # default stays False so ungated/public engines never fetch internal
156 # URLs. It applies (to that validation and to the download
157 # pipeline) only when the run policy imposes no host-class
158 # restriction,
159 # i.e. without a run context or under ``unprotected``
160 # (``_effective_allow_private_ips``).
161 # Link-local and the cloud-metadata literals in
162 # ALWAYS_BLOCKED_METADATA_IPS stay blocked regardless (issue #2477).
163 self.allow_private_ips = allow_private_ips
164 # Optional ``(done, total)`` liveness hook for the bulk page
165 # fetch below (see ``batch_fetch_and_extract``). Threaded in by
166 # the pipeline strategy (``AdvancedSearchSystem`` → ``get_search``)
167 # and by agent search tools so a minutes-long fetch emits progress
168 # instead of silence; ``None`` disables it with zero overhead.
169 self.progress_callback = progress_callback
170 # Set by the factory when the parent engine is gated against a
171 # specific scope; used to evaluate per-URL fetches below.
172 self.egress_context = egress_context
173 self.content_cache = content_cache
174 # One operator-visible line per wrapper when the grant is set but
175 # the run scope keeps it from applying
176 # (``_effective_allow_private_ips``).
177 self._grant_withheld_logged = False
179 def _resolve_cache(self) -> Optional[Any]:
180 if self.content_cache is not None:
181 return self.content_cache
182 try:
183 from ...utilities.research_content_cache import get_current_cache
184 except Exception: # pragma: no cover - defensive
185 return None
186 try:
187 return get_current_cache()
188 except Exception: # pragma: no cover - defensive
189 return None
191 def set_fetch_progress_callback(self, hook) -> None:
192 """Late-bind the ``(done, total)`` bulk-fetch hook.
194 Used by ``AdvancedSearchSystem.set_progress_callback`` when the UI
195 callback arrives after engine construction.
196 """
197 self.progress_callback = hook
199 @property
200 def last_search_failure(self) -> SearchFailure | None:
201 """Expose provider failures through the optional full-content wrapper."""
202 return engine_failure(self.web_search)
204 def _effective_allow_private_ips(self) -> bool:
205 """The private result-fetch grant as it applies to this run.
207 Used for BOTH the SSRF validation of result URLs in ``run`` /
208 ``_get_full_content`` and the download pipeline. The pipeline
209 checks each redirect and pinned address against the run scope. The
210 private-result grant remains limited to no run context at all,
211 or a scope in ``_SCOPES_WITHOUT_HOST_CLASS_RESTRICTION``
212 (``unprotected``). Under every other scope the grant is withheld
213 from both, so a private result URL is refused by ``validate_url``
214 exactly as if the operator had not set the opt-in. That refusal
215 happens before ``evaluate_url`` runs, so it does not consume the
216 run's ``MAX_DENIED_FETCHES_PER_RUN`` quota. Hosts the SSRF validator
217 admits but the policy cannot classify public (including benchmark
218 and documentation hosts) are refused uncounted by ``_refuse_private_uncounted`` instead, under
219 the same scopes (every scope but ``unprotected``).
220 """
221 if not self.allow_private_ips:
222 return False
223 egress_ctx = self.egress_context
224 if egress_ctx is None:
225 return True
226 scope = getattr(egress_ctx, "scope", None)
227 if scope in _SCOPES_WITHOUT_HOST_CLASS_RESTRICTION:
228 return True
229 if not getattr(self, "_grant_withheld_logged", False):
230 self._grant_withheld_logged = True
231 scope_name = getattr(scope, "value", scope)
232 logger.info(
233 "full_search: private result-fetch grant "
234 "(LDR_SEARCH_ALLOW_PRIVATE_RESULT_FETCH) not applied under "
235 f"egress scope {scope_name}; it takes effect only under the "
236 "unprotected scope, so private result URLs stay blocked"
237 )
238 return False
240 @staticmethod
241 def _refuse_private_uncounted(url: str, egress_ctx: Any) -> bool:
242 """Drop a result URL whose host the policy cannot classify public,
243 uncounted, under every scope that withholds the private-fetch grant.
245 Private result URLs stay blocked under every scope but
246 ``unprotected`` (``_effective_allow_private_ips``). ``validate_url``
247 refuses most of them, but its ``PRIVATE_IP_RANGES`` is narrower than
248 the policy's classifier: it also admits benchmark, documentation and
249 other special-purpose address ranges. ``evaluate_url`` counts
250 denials under every scope, so a page of these URLs could exhaust
251 the run's ``MAX_DENIED_FETCHES_PER_RUN`` quota and block later public
252 fetches. It may admit private hosts under ``strict`` /
253 ``private_only`` / ``both``. So they are refused
254 here, before ``evaluate_url``, through the policy's own
255 non-counting ``host_not_classified_public`` (same host
256 classification as ``evaluate_url``), under every scope except the
257 ones in ``_SCOPES_WITHOUT_HOST_CLASS_RESTRICTION``; an unknown scope
258 fails closed. A URL with an empty host is refused too; an unresolvable
259 host or a DNS timeout is classified public (the classifier's fail-safe)
260 and is left to ``validate_url`` and ``evaluate_url``.
261 This uncounted check covers the result URL. Static redirects and
262 pinned connect addresses receive the transport's SSRF and run-scope
263 checks. Browser fallback is disabled under restricted scopes; when
264 available under Both or Unprotected, it retains SSRF checks.
265 """
266 scope = getattr(egress_ctx, "scope", None)
267 if scope in _SCOPES_WITHOUT_HOST_CLASS_RESTRICTION:
268 return False
269 from ...security.egress.policy import host_not_classified_public
271 if not host_not_classified_public(url, egress_ctx):
272 return False
273 scope_name = getattr(scope, "value", scope)
274 logger.bind(policy_audit=True).warning(
275 "full_search URL denied by egress policy (non-public result "
276 "host, not counted toward the denied-fetch quota)",
277 url=redact_url_for_log(url),
278 scope=scope_name,
279 reason=f"nonpublic_result_host_{scope_name}",
280 )
281 return True
283 def _url_quality_prompt(self, results: List[Dict], query: str) -> str:
284 """Build the URL-quality prompt shared by the sync and async paths."""
285 now = datetime.now(UTC)
286 current_time = now.strftime("%Y-%m-%d")
287 return f"""ONLY Return a JSON array. The response contains no letters. Evaluate these URLs for:
288 1. Timeliness (today: {current_time})
289 2. Factual accuracy (cross-reference major claims)
290 3. Source reliability (prefer official company websites, established news outlets)
291 4. Direct relevance to query: {query}
293 URLs to evaluate:
294 {results}
296 Return a JSON array of indices (0-based) for sources that meet ALL criteria.
297 ONLY Return a JSON array of indices (0-based) and nothing else. No letters.
298 Example response: \n[0, 2, 4]\n\n"""
300 def _keep_selected(self, results: List[Dict], response: Any) -> List[Dict]:
301 """Turn the LLM's index array into the filtered result list."""
302 response_text = get_llm_response_text(response)
303 good_indices = extract_json(response_text, expected_type=list)
305 if good_indices is None:
306 good_indices = []
308 return [r for i, r in enumerate(results) if i in good_indices]
310 def _url_filter_error_message(self, error: Exception) -> str:
311 """Build the scrubbed log message for an LLM-side filter failure."""
312 safe_msg = scrub_error(error)
313 return f"URL filtering error ({type(error).__name__}): {safe_msg}"
315 def _unfiltered_fallback(self, results: List[Dict]) -> List[Dict]:
316 """Fall back to unfiltered results after an LLM-side filter failure.
318 The ``logger.exception(...)`` call itself stays in each
319 ``except Exception as e:`` block below (sync and async) rather than
320 here, so the caught exception stays visible to
321 ``.pre-commit-hooks/check-sensitive-logging.py`` — it tracks
322 ``except ... as name`` bindings, and a helper taking the exception
323 as a plain parameter falls outside what it can see.
324 """
325 logger.warning(
326 "URL quality filter unavailable — returning {} unfiltered "
327 "results as fallback",
328 len(results),
329 )
330 return results # Fall back to original results on LLM error
332 def check_urls(self, results: List[Dict], query: str) -> List[Dict]:
333 """Filter results down to the URLs the LLM judges usable.
335 Stays on the synchronous LangChain API: FullSearchResults runs inside
336 per-run research threads with no event loop, and langchain's async
337 httpx client is process-cached and loop-bound, so driving the async
338 core on a throwaway per-call loop would break or re-send every call
339 after the first (see #6293).
340 """
341 if not results:
342 return results
344 prompt = self._url_quality_prompt(results, query)
346 try:
347 if self.llm is None:
348 return results
349 return self._keep_selected(
350 results, invoke_llm_sync(self.llm, prompt)
351 )
352 except PolicyDeniedError:
353 # The URL-quality LLM was denied by egress policy (e.g. a cloud
354 # LLM under require_local / PRIVATE_ONLY). Do NOT fall through to
355 # the unfiltered-results fallback below — that would let the run
356 # proceed despite the user's policy refusing the LLM. Fail closed.
357 raise
358 except Exception as e:
359 safe_msg = self._url_filter_error_message(e)
360 logger.exception(safe_msg)
361 return self._unfiltered_fallback(results)
363 async def _check_urls_async(
364 self, results: List[Dict], query: str
365 ) -> List[Dict]:
366 """Async counterpart of :meth:`check_urls`.
368 Additive: no production caller awaits it yet. It is intended to be
369 submitted to the process's one shared event loop via
370 ``asyncio.run_coroutine_threadsafe`` (ADR-0013,
371 docs/decisions/0013-research-runs-keep-daemon-thread.md); which
372 thread hosts that loop is not decided yet (#6467).
373 :meth:`check_urls` remains the canonical production path on the
374 research thread, and this must never go through ``asyncio.run`` or
375 a per-thread loop (#6293).
377 Not loop-safe yet, for three reasons: ``TokenCountingCallback``
378 writes call-usage rows to the user's encrypted database from the
379 loop's default executor rather than the daemon thread (rule 5;
380 #6294 is related); the egress audit context is thread-local and
381 must be passed in and re-armed on the loop side; and the submit
382 helper must run the coroutine in an explicit minimal context so it
383 does not inherit the daemon thread's ambient search context,
384 including the password (rule 4).
385 """
386 if not results: 386 ↛ 387line 386 didn't jump to line 387 because the condition on line 386 was never true
387 return results
389 prompt = self._url_quality_prompt(results, query)
391 try:
392 if self.llm is None: 392 ↛ 393line 392 didn't jump to line 393 because the condition on line 392 was never true
393 return results
394 return self._keep_selected(results, await self.llm.ainvoke(prompt))
395 except PolicyDeniedError:
396 # Fail closed exactly like the sync path: an egress-policy denial
397 # must not fall through to the unfiltered-results fallback.
398 raise
399 except Exception as e:
400 safe_msg = self._url_filter_error_message(e)
401 logger.exception(safe_msg)
402 return self._unfiltered_fallback(results)
404 def run(
405 self, query: str, research_context: Dict[str, Any] | None = None
406 ) -> List[Dict[str, Any]]:
407 # ``research_context`` is accepted for signature parity with
408 # ``BaseSearchEngine.run`` and its other overrides (e.g. SearXNG,
409 # arXiv), which callers such as ``FocusedIterationStrategy``
410 # invoke positionally by keyword. This engine derives everything
411 # it needs from ``query`` and its own settings, so the value is
412 # deliberately unused.
413 # Step 1: Get search results
414 search_results = self.web_search.invoke(query)
415 if not isinstance(search_results, list):
416 raise ValueError("Expected the search results in list format.")
418 # Step 2: Filter URLs using LLM if enabled
419 should_check_urls = resolve_boolean_setting(
420 self.settings_snapshot,
421 "search.quality_check_urls",
422 default=QUALITY_CHECK_DDG_URLS,
423 )
424 if should_check_urls and self.llm is not None:
425 logger.info(
426 "Evaluating search result URLs with LLM before fetching full content..."
427 )
428 filtered_results = self.check_urls(search_results, query)
429 else:
430 filtered_results = search_results
432 # Extract URLs from filtered results
433 urls = [
434 result.get("link")
435 for result in filtered_results
436 if result.get("link")
437 ]
439 if not urls:
440 logger.error("\n === NO VALID LINKS ===\n")
441 return []
443 # SSRF-validate + egress-scope-validate each URL. SSRF only checks
444 # IP class; scope enforcement is the separate axis that blocks
445 # public hosts under PRIVATE_ONLY / STRICT.
446 evaluate_url_fn = None
447 egress_ctx = self.egress_context
448 if egress_ctx is not None:
449 from ...security.egress.policy import evaluate_url as _ev_url
451 evaluate_url_fn = _ev_url
452 allow_private = self._effective_allow_private_ips()
453 safe_urls: List[str] = []
454 for url in urls:
455 if url is None: 455 ↛ 456line 455 didn't jump to line 456 because the condition on line 455 was never true
456 continue
457 # block_link_local=True is the carve-out inside the private-IP
458 # grant: the rest of the link-local range hosts provider metadata
459 # beyond ALWAYS_BLOCKED_METADATA_IPS (Scaleway's endpoint, for
460 # one) and no legitimate result page lives there. No effect
461 # without the grant, where link-local is already blocked.
462 if not validate_url(
463 url,
464 allow_private_ips=allow_private,
465 block_link_local=True,
466 ):
467 logger.warning(
468 "SSRF validation blocked URL from full content fetch: "
469 f"{redact_url_for_log(url)}."
470 )
471 continue
472 if egress_ctx is not None and self._refuse_private_uncounted(
473 url, egress_ctx
474 ):
475 continue
476 if evaluate_url_fn is not None and egress_ctx is not None:
477 url_decision = evaluate_url_fn(url, egress_ctx)
478 if not url_decision.allowed:
479 logger.bind(policy_audit=True).warning(
480 "full_search URL denied by egress policy",
481 url=redact_url_for_log(url),
482 scope=egress_ctx.scope.value,
483 reason=url_decision.reason,
484 )
485 continue
486 safe_urls.append(url)
488 if not safe_urls:
489 logger.warning(
490 "All URLs were blocked by SSRF validation — returning results "
491 "without full content. This can happen when search results "
492 "point to internal/private network addresses."
493 )
494 for result in filtered_results:
495 result["full_content"] = None
496 return filtered_results
498 # Fetch and extract all pages — specialized downloaders (arXiv,
499 # PubMed, etc.) are tried first, with HTML crawling as fallback for
500 # every type but arXiv paper URLs, which their downloader owns
501 # outright: for those a failure is terminal and yields no content.
502 # Other arXiv-host pages are not owned and do crawl.
503 logger.info(
504 f"Full search: fetching and extracting full content for {len(safe_urls)} pages..."
505 )
506 url_to_content = batch_fetch_and_extract(
507 safe_urls,
508 language=self.language,
509 enable_js_rendering=_read_js_rendering_setting(
510 self.settings_snapshot
511 ),
512 content_cache=self._resolve_cache(),
513 allow_private_ips=allow_private,
514 block_link_local=True,
515 egress_context=egress_ctx,
516 progress_callback=self.progress_callback,
517 )
519 nr_full_text = sum(1 for v in url_to_content.values() if v)
520 for result in filtered_results:
521 link = result.get("link")
522 result["full_content"] = url_to_content.get(link) if link else None
523 result["full_content_is_complete"] = bool(result["full_content"])
525 logger.info(f"Full search: retrieved content from {nr_full_text} pages")
526 return filtered_results
528 def _get_full_content(
529 self, relevant_items: List[Dict[str, Any]]
530 ) -> List[Dict[str, Any]]:
531 """Fetch and attach full content to an existing list of items."""
532 evaluate_url_fn = None
533 egress_ctx = self.egress_context
534 if egress_ctx is not None:
535 from ...security.egress.policy import evaluate_url as _ev_url
537 evaluate_url_fn = _ev_url
538 allow_private = self._effective_allow_private_ips()
539 urls: List[str] = []
540 for item in relevant_items:
541 link = item.get("link")
542 if link is None:
543 continue
544 if not validate_url(
545 link,
546 allow_private_ips=allow_private,
547 block_link_local=True,
548 ):
549 logger.warning(
550 "SSRF validation blocked URL from full content fetch: "
551 f"{redact_url_for_log(link)}."
552 )
553 continue
554 if egress_ctx is not None and self._refuse_private_uncounted(
555 link, egress_ctx
556 ):
557 continue
558 if evaluate_url_fn is not None and egress_ctx is not None:
559 url_decision = evaluate_url_fn(link, egress_ctx)
560 if not url_decision.allowed:
561 logger.bind(policy_audit=True).warning(
562 "full_search _get_full_content URL denied by egress policy",
563 url=redact_url_for_log(link),
564 scope=egress_ctx.scope.value,
565 reason=url_decision.reason,
566 )
567 continue
568 urls.append(link)
570 if not urls:
571 for item in relevant_items:
572 item["full_content"] = None
573 return relevant_items
575 try:
576 url_to_content = batch_fetch_and_extract(
577 urls,
578 language=self.language,
579 enable_js_rendering=_read_js_rendering_setting(
580 self.settings_snapshot
581 ),
582 content_cache=self._resolve_cache(),
583 allow_private_ips=allow_private,
584 block_link_local=True,
585 egress_context=egress_ctx,
586 progress_callback=self.progress_callback,
587 )
588 except PolicyDeniedError:
589 raise
590 except Exception as e:
591 safe_msg = scrub_error(e)
592 logger.exception(
593 f"Error fetching full content ({type(e).__name__}): {safe_msg}"
594 )
595 for item in relevant_items:
596 item["full_content"] = None
597 return relevant_items
599 for item in relevant_items:
600 link = item.get("link")
601 item["full_content"] = url_to_content.get(link) if link else None
602 item["full_content_is_complete"] = bool(item["full_content"])
604 return relevant_items
606 def close(self) -> None:
607 """Close the underlying search engine and any held resources."""
608 from ...utilities.resource_utils import safe_close
610 if hasattr(self, "web_search") and self.web_search is not None: 610 ↛ exitline 610 didn't return from function 'close' because the condition on line 610 was always true
611 safe_close(
612 self.web_search,
613 "wrapped web search engine",
614 closing_optional=True,
615 )
617 def __enter__(self):
618 """Support context manager usage."""
619 return self
621 def __exit__(self, exc_type, exc_val, exc_tb):
622 """Cleanup on context exit."""
623 self.close()
624 return False
626 def invoke(self, query: str) -> Any:
627 return self.run(query)
629 def __call__(self, query: str) -> Any:
630 return self.invoke(query)