Coverage for src/local_deep_research/web_search_engines/search_engine_base.py: 95%
670 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1import json
2import time
3from abc import ABC, abstractmethod
4from contextvars import ContextVar
5from dataclasses import dataclass
6from typing import (
7 TYPE_CHECKING,
8 Any,
9 Callable,
10 Dict,
11 List,
12 Optional,
13 Set,
14 Union,
15)
16from urllib.parse import urlparse
18from langchain_core.language_models import BaseLLM
19from ..security.secure_logging import logger
20from tenacity import (
21 retry,
22 retry_if_exception_type,
23 stop_after_attempt,
24)
25from tenacity.wait import wait_base
27from ..advanced_search_system.filters.base_filter import BaseFilter
28from ..utilities.type_utils import _is_strict_scope, unwrap_setting
29from ..config.constants import DEFAULT_MAX_FILTERED_RESULTS
30from ..config.thread_settings import get_setting_from_snapshot
31from ..security.log_sanitizer import sanitize_error_message, scrub_error
32from ..security.egress.classification import Exposure, Sensitivity
33from ..utilities.thread_context import clear_search_context, set_search_context
35from .rate_limiting import RateLimitError, get_tracker
36from .engine_availability import (
37 SearchFailure,
38 failure_from_exception,
39 is_query_error,
40 retry_after_from_error,
41)
43if TYPE_CHECKING:
44 from ..advanced_search_system.filters.journal_reputation_filter import (
45 JournalReputationFilter,
46 )
49# Common placeholder values that should never be treated as a real API key.
50# Three call sites previously had inconsistent subsets of this list, which
51# let invalid placeholders silently through the production path in
52# search_engines_config.py. Centralizing here so all three stay in sync.
53#
54# All entries are lowercased — comparison is case-insensitive (see
55# ``_is_api_key_placeholder``). Substring patterns like ``endswith("_API_KEY")``
56# are intentionally NOT used: they silently rejected legitimate keys that
57# happened to contain those substrings (e.g. ``sk-real-secret_API_KEY``).
58# Users who paste an env-var name like ``BRAVE_API_KEY`` instead of its value
59# will get a clear API-call-time auth failure instead.
60API_KEY_PLACEHOLDERS = frozenset(
61 {
62 "",
63 "none",
64 "null",
65 "placeholder",
66 # Underscore forms
67 "your_api_key_here",
68 "your_api_key",
69 "api_key",
70 "insert_api_key_here",
71 "insert_api_key",
72 # Hyphenated forms shipped in ``src/local_deep_research/defaults/.env.template``
73 # or ``docs/env_configuration.md`` as the literal value users are told to
74 # copy-paste. Each must be rejected verbatim. Drift is enforced by
75 # ``TestEnvTemplatePlaceholdersCovered`` and ``TestEnvDocsPlaceholdersCovered``.
76 "your-api-key-here",
77 "your-anthropic-key-here",
78 "your-openai-key-here",
79 "your-openrouter-key-here",
80 "your-google-api-key-here",
81 "your-brave-key-here",
82 "your-serpapi-key-here",
83 "your-tavily-key-here",
84 # Defensive extras — plausible variations users might type but not in
85 # any template/docs. Listed explicitly so they don't silently accumulate.
86 "your-api-key", # without -here suffix
87 "your-key-here", # without api- prefix
88 "your-azure-key", # Azure provider variant
89 }
90)
92# Canonical short codes for time-restricted search (the global
93# ``search.time_period`` setting minus "all", which means "no filter" and is
94# expressed by omitting the engine's recency param entirely). Engines that
95# forward a recency filter to their API (brave, serpapi, tavily) validate
96# against this shared set so the accepted codes can't drift apart.
97VALID_TIME_PERIODS = frozenset({"d", "w", "m", "y"})
100def _is_api_key_placeholder(api_key: Optional[str]) -> bool:
101 """Return True if ``api_key`` looks like a placeholder, not a real key.
103 Catches:
104 - Exact-match placeholders, case-insensitively (see ``API_KEY_PLACEHOLDERS``)
105 - Angle-bracket templates: ``<...>`` (e.g. ``<your_api_key>``)
106 - Shell-variable templates: ``${...}`` (e.g. ``${BRAVE_API_KEY}``)
107 """
108 if not api_key:
109 return True
110 api_key = api_key.strip().lower()
111 if api_key in API_KEY_PLACEHOLDERS:
112 return True
113 if api_key.startswith("<") and api_key.endswith(">"):
114 return True
115 if api_key.startswith("${") and api_key.endswith("}"):
116 return True
117 return False
120@dataclass
121class _SearchRunState:
122 """Outcomes belonging to one search invocation, including its retries."""
124 search_failed: bool = False
125 rate_limited_after_previews: bool = False
128class AdaptiveWait(wait_base):
129 """Custom wait strategy that uses adaptive rate limiting."""
131 def __init__(self, get_wait_func):
132 self.get_wait_func = get_wait_func
134 def __call__(self, retry_state):
135 return self.get_wait_func()
138def metrics_engine_name(engine: Any) -> str:
139 """The engine name ``SearchTracker`` records for ``engine``'s searches."""
140 return type(engine).__name__.replace("SearchEngine", "").lower()
143class BaseSearchEngine(ABC):
144 """
145 Abstract base class for search engines with two-phase retrieval capability.
146 Handles common parameters and implements the two-phase search approach.
148 Subclass contract for ``__init__``
149 ---------------------------------
150 Concrete engines should forward ``settings_snapshot`` and
151 ``programmatic_mode`` to ``super().__init__`` so the base class can wire
152 them up correctly. The cleanest way is to declare ``**kwargs`` and pass
153 it along::
155 def __init__(self, max_results=10, *, my_param=None, **kwargs):
156 super().__init__(max_results=max_results, **kwargs)
157 self.my_param = my_param
159 Many existing engines accept ``**kwargs`` but don't forward — that
160 silently drops ``programmatic_mode`` (and used to bind the wrong rate
161 tracker). The factory has a post-construction safety net that calls
162 ``_configure_programmatic_mode`` when the engine's mode doesn't match
163 what the caller asked for, but new engines should not rely on it: the
164 safety net only covers ``programmatic_mode``, not ``settings_snapshot``
165 or other base kwargs.
166 """
168 # Class attribute to indicate if this engine searches public internet sources
169 # Should be overridden by subclasses - defaults to False for safety
170 is_public = False
172 last_search_failure: SearchFailure | None = None
173 # True when the provider answers an invalid ``self.api_key`` with a plain
174 # HTTP 403 that carries no authentication hint. A 403 on a request sent
175 # with a configured key then disables the engine for the run instead of
176 # cooling it down again and again.
177 bare_403_means_bad_api_key = False
179 # Class attribute to indicate if this is a generic search engine (vs specialized)
180 # Generic engines are general web search (Google, Bing, etc) vs specialized (arXiv, PubMed).
181 # Note: generic does NOT imply good native ranking — see is_lexical.
182 is_generic = False
184 # Class attribute to indicate if this is a scientific/academic search engine
185 # Scientific engines include arXiv, PubMed, Semantic Scholar, etc.
186 is_scientific = False
188 # Class attribute to indicate if this is a local RAG/document search engine
189 # Local engines search private document collections stored locally
190 is_local = False
192 # Egress data classification (ADR-0007). Two orthogonal axes, declared
193 # explicitly per engine alongside is_public/is_local. The defaults fail
194 # closed: an engine that forgets to classify itself is treated as the most
195 # restrictive quadrant (sensitive data + exposing sink → usable only by
196 # itself). Engines whose label depends on configuration (a collection's
197 # is_public flag, a self-hosted store's URL) are refined by the classifier
198 # at run time; these constants are the static fallback.
199 egress_sensitivity = Sensitivity.SENSITIVE
200 egress_exposure = Exposure.EXPOSING
202 # Whether full-content fetches of this engine's RESULT URLs may reach
203 # private / loopback hosts. The search-engine factory reads it when it
204 # builds the FullSearchResults wrapper (the only full-content path) and
205 # hands it over as the wrapper's ``allow_private_ips``. Default False:
206 # an engine that holds such a grant resolves it itself in ``__init__``
207 # (SearXNG: instance approval AND the env-only
208 # ``search.allow_private_result_fetch`` opt-in); every other engine keeps
209 # private result URLs blocked (issue #2477). The wrapper applies the
210 # grant only under the run's ``unprotected`` egress scope.
211 allow_private_result_fetch = False
213 # Class attribute to indicate if this is a news search engine
214 # News engines specialize in news articles and current events
215 is_news = False
217 # Class attribute to indicate if this is a code search engine
218 # Code engines specialize in searching code repositories
219 is_code = False
221 # Class attribute to indicate if this is a book/literature search engine
222 # Book engines search libraries and literary archives
223 is_books = False
225 # Classification: does this engine use lexical/keyword-based search?
226 # Lexical engines (arXiv, PubMed, Wikipedia, Mojeek, etc.) match results by
227 # keywords without ML-based ranking. This is an informational flag that can
228 # drive multiple behaviors (query optimization, result deduplication, UI hints).
229 # For LLM relevance filtering specifically, see needs_llm_relevance_filter.
230 is_lexical = False
232 # Behavioral: should the factory auto-enable LLM relevance filtering?
233 # When True, the factory sets enable_llm_relevance_filter=True on the engine
234 # instance, causing _filter_for_relevance() to run after previews are fetched.
235 # Typically set alongside is_lexical=True, but can be set independently —
236 # e.g. a non-lexical engine with noisy results could opt in.
237 needs_llm_relevance_filter = False
239 # Tuning for the LLM relevance filter (only applies when the filter
240 # is active for this engine).
241 #
242 # relevance_filter_batch_size: split previews into chunks of this many
243 # before sending to the LLM. Smaller batches are faster per call and
244 # more reliable on weaker models which struggle with many indices in
245 # one context. None or 0 = single-call mode (no batching).
246 #
247 # relevance_filter_max_parallel_batches: number of batches to dispatch
248 # concurrently against the LLM. 1 = sequential. Most providers handle
249 # parallel requests fine (Ollama with OLLAMA_NUM_PARALLEL>1, OpenAI,
250 # Anthropic).
251 relevance_filter_batch_size: Optional[int] = 5
252 relevance_filter_max_parallel_batches: int = 10
254 # Class attribute for rate limit detection patterns
255 # Subclasses can override to add engine-specific patterns
256 rate_limit_patterns: Set[str] = {
257 "rate limit",
258 "rate_limit",
259 "ratelimit",
260 "too many requests",
261 "throttl",
262 "quota exceeded",
263 "quota_exceeded",
264 "limit exceeded",
265 "request limit",
266 "api limit",
267 "usage limit",
268 }
270 # Instance-attribute names holding credential values to redact from error
271 # messages/logs via _scrub_error(). Subclasses override when they store
272 # secrets under different names (e.g. Elasticsearch: _api_key/_password).
273 # Centralizing the list keeps every _scrub_error() call site uniform and
274 # prevents the per-site drift that previously dropped a secret.
275 # ``_rejected_api_key`` is in the default because the OpenAlex DOI
276 # enrichment pass can store a refused key there on *any* scientific
277 # engine (arxiv, semantic_scholar, …), not just OpenAlexSearchEngine.
278 # ``_scrub_error`` reads it through ``getattr(..., None)``, so engines
279 # that never see a rejection are unaffected.
280 _secret_attrs: tuple[str, ...] = ("api_key", "_rejected_api_key")
282 # Class-level defaults so every engine has them without touching
283 # __init__ (and so a ``__new__``-built instance degrades rather than
284 # raising). ``_openalex_key_rejected`` latches "OpenAlex refused this
285 # credential" for the instance's lifetime: dropping the key from the
286 # engine's own attributes is not enough, because
287 # ``_resolve_openalex_enrichment_key`` would otherwise re-read the
288 # very same key out of the settings snapshot it came from and send it
289 # again on the enrichment request of the same ``run()``.
290 _openalex_key_rejected: bool = False
291 # The refused literal, kept only so it stays in the redaction set
292 # (see ``_secret_attrs``) after the live attribute is cleared.
293 _rejected_api_key: Optional[str] = None
295 def _new_search_run_state(self) -> _SearchRunState:
296 """Allow adapters to keep additional retry state in the same scope."""
297 return _SearchRunState()
299 def _search_run_state_var(self) -> ContextVar[Optional[_SearchRunState]]:
300 # Normally initialized in __init__; retain support for bare instances
301 # used to call individual adapter helpers without constructing an engine.
302 state_var: ContextVar[Optional[_SearchRunState]] | None = getattr(
303 self, "_search_run_state", None
304 )
305 if state_var is None:
306 state_var = ContextVar("search_run_state", default=None)
307 self._search_run_state = state_var
308 return state_var
310 def _get_search_run_state(self) -> _SearchRunState:
311 state_var = self._search_run_state_var()
312 state = state_var.get()
313 if state is None:
314 # Direct calls to private adapter helpers can share retry state in
315 # this context. Public run() always starts a fresh invocation.
316 state = self._new_search_run_state()
317 state_var.set(state)
318 return state
320 @property
321 def _search_failed(self) -> bool:
322 """An absorbed preview failure in the current search attempt."""
323 return self._get_search_run_state().search_failed
325 @_search_failed.setter
326 def _search_failed(self, value: bool) -> None:
327 self._get_search_run_state().search_failed = value
329 @property
330 def _rate_limited_after_previews(self) -> bool:
331 """An absorbed full-content HTTP 429 in the current search attempt."""
332 return self._get_search_run_state().rate_limited_after_previews
334 @_rate_limited_after_previews.setter
335 def _rate_limited_after_previews(self, value: bool) -> None:
336 self._get_search_run_state().rate_limited_after_previews = value
338 def _note_openalex_key_rejected(
339 self, api_key: Optional[str] = None
340 ) -> None:
341 """Latch that OpenAlex refused *api_key*, for this engine's life.
343 Called by every OpenAlex path that can learn a key is bad: the
344 search engine's own ``_drop_rejected_api_key`` and the DOI
345 enrichment pass's ``on_key_rejected`` callback. Once set,
346 ``_resolve_openalex_enrichment_key`` returns ``None``, so no
347 path on this object sends the refused credential a second time.
349 ``or self._rejected_api_key`` keeps the stored literal: a second
350 call (two threads driving one engine) must not overwrite it with
351 ``None`` and silently remove it from the redaction set.
352 """
353 self._openalex_key_rejected = True
354 self._rejected_api_key = api_key or self._rejected_api_key
356 @staticmethod
357 def _ensure_list(value, *, default=None):
358 """Normalize a value that should be a list.
360 Handles JSON-encoded strings, comma-separated strings, and
361 already-parsed lists. Returns *default* (empty list when not
362 supplied) for ``None`` or empty/unparseable input.
363 """
364 if default is None:
365 default = []
366 if value is None:
367 return default
368 if isinstance(value, list):
369 return value
370 if isinstance(value, str):
371 stripped = value.strip()
372 if not stripped:
373 return default
374 if stripped.startswith("["):
375 try:
376 parsed = json.loads(stripped)
377 if isinstance(parsed, list): 377 ↛ 381line 377 didn't jump to line 381 because the condition on line 377 was always true
378 return [str(item) for item in parsed]
379 except (json.JSONDecodeError, ValueError, RecursionError):
380 pass
381 return [
382 item.strip() for item in stripped.split(",") if item.strip()
383 ]
384 return default
386 @classmethod
387 def _load_engine_class(cls, name: str, config: Dict[str, Any]):
388 """
389 Helper method to load an engine class dynamically.
391 Args:
392 name: Engine name
393 config: Engine configuration dict with module_path and class_name
395 Returns:
396 Tuple of (success: bool, engine_class or None, error_msg or None)
397 """
398 from ..security.module_whitelist import (
399 ModuleNotAllowedError,
400 get_safe_module_class,
401 )
403 try:
404 module_path = config.get("module_path")
405 class_name = config.get("class_name")
407 if not module_path or not class_name:
408 return (
409 False,
410 None,
411 f"Missing module_path or class_name for {name}",
412 )
414 # Use whitelist-validated safe import
415 engine_class = get_safe_module_class(module_path, class_name)
417 return True, engine_class, None
419 except ModuleNotAllowedError as e:
420 return (
421 False,
422 None,
423 f"Security error loading engine class for {name}: "
424 f"{scrub_error(e)}",
425 )
426 except Exception as e:
427 return (
428 False,
429 None,
430 f"Could not load engine class for {name}: {scrub_error(e)}",
431 )
433 @classmethod
434 def is_available(
435 cls, settings_snapshot: Optional[Dict[str, Any]] = None
436 ) -> bool:
437 """Runtime availability probe used by ``list_eligible_engine_configs``.
439 Subclasses whose ``__init__`` performs a network probe that can
440 hard-fail (e.g. Elasticsearch's ``client.info()`` raises
441 ``ConnectionError`` when the cluster is unreachable) should override
442 this to fail closed: returning ``False`` causes the engine to be
443 omitted from specialized agent tool surfaces and the auto-search
444 candidate pool, so those paths do not advertise or instantiate a
445 broken engine. The primary generic ``web_search`` tool and direct
446 factory callers can still attempt construction and surface the
447 underlying configuration or connection error.
449 Default is ``True`` — engines that don't probe at init stay
450 eligible. This is the opposite default of ``BaseLLMProvider`` because
451 the vast majority of search engines here are pure REST clients with
452 lazy connection errors; changing the default would force every
453 subclass to opt in to ``True`` for no benefit.
454 """
455 return True
457 def __init__(
458 self,
459 llm: Optional[BaseLLM] = None,
460 max_filtered_results: Optional[int] = None,
461 max_results: Optional[int] = 10, # Default value if not provided
462 preview_filters: List[BaseFilter] | None = None,
463 content_filters: List[BaseFilter] | None = None,
464 search_snippets_only: bool = True, # New parameter with default
465 include_full_content: bool = False,
466 settings_snapshot: Optional[Dict[str, Any]] = None,
467 programmatic_mode: bool = False,
468 content_cache: Any = None,
469 progress_callback=None,
470 **kwargs,
471 ):
472 """
473 Initialize the search engine with common parameters.
475 Args:
476 llm: Optional language model for relevance filtering
477 max_filtered_results: Maximum number of results to keep after filtering
478 max_results: Maximum number of search results to return
479 preview_filters: Filters that will be applied to all previews
480 produced by the search engine, before relevancy checks.
481 content_filters: Filters that will be applied to the full content
482 produced by the search engine, after relevancy checks.
483 search_snippets_only: Whether to return only snippets or full content
484 include_full_content: Whether to use FullSearchResults for full webpage content
485 settings_snapshot: Settings snapshot for configuration
486 programmatic_mode: If True, disables database operations and uses memory-only tracking
487 content_cache: Optional shared ``ResearchContentCache`` for the
488 whole research run. When None, the cache bound to the
489 current research (if any) is used, so repeat URLs across
490 the quick-report phase, detailed subsections, lead agent
491 and subagents are served without re-fetching on the spot.
492 progress_callback: Optional ``(done, total)`` bulk-fetch liveness
493 hook forwarded to ``FullSearchResults`` via
494 ``_init_full_search`` / ``set_fetch_progress_callback``
495 (pipeline-strategy fetch progress, #7193).
496 **kwargs: Additional engine-specific parameters
497 """
498 self._search_run_state = ContextVar("search_run_state", default=None)
499 if max_filtered_results is None:
500 max_filtered_results = DEFAULT_MAX_FILTERED_RESULTS
501 if max_results is None:
502 max_results = 10
503 self._preview_filters: List[BaseFilter] = (
504 preview_filters if preview_filters is not None else []
505 )
506 self._content_filters: List[BaseFilter] = (
507 content_filters if content_filters is not None else []
508 )
510 self.llm = llm # LLM for relevance filtering
511 self._max_filtered_results = int(
512 max_filtered_results
513 ) # Ensure it's an integer
514 self._max_results = max(
515 1, int(max_results)
516 ) # Ensure it's a positive integer
517 self.search_snippets_only = search_snippets_only # Store the setting
518 self.include_full_content = include_full_content
519 self.settings_snapshot = (
520 settings_snapshot or {}
521 ) # Store settings snapshot
522 # Single shared cache for the whole research run. Also honoured
523 # from **kwargs for engines whose __init__ forwards extras.
524 if content_cache is None:
525 content_cache = kwargs.get("content_cache")
526 self.content_cache = content_cache
527 # Bulk-fetch liveness hook (``(done, total)``) for full-search
528 # progress. Set at construction via the factory or later via
529 # ``set_fetch_progress_callback`` (e.g. when
530 # ``AdvancedSearchSystem.set_progress_callback`` arrives after
531 # engine creation). Forwarded to ``FullSearchResults``.
532 self._fetch_progress_callback = progress_callback
534 self.engine_type = self.__class__.__name__
535 self._engine_name: str = "" # set by the factory after construction
536 # The snapshot dict the runtime egress backstop last verified
537 # against (identity comparison; holding the reference keeps the
538 # memo valid — a verification re-runs when the caller assigns a
539 # NEW snapshot) plus the policy-relevant VALUES it carried at
540 # verification time, so an in-place mutation of the scope or
541 # primary key on the same dict is also caught.
542 self._egress_verified_snapshot: Optional[Dict[str, Any]] = None
543 self._egress_verified_policy_key: Optional[tuple] = None
544 self._egress_skip_warned = False
545 self._configure_programmatic_mode(programmatic_mode)
546 self._last_wait_time = (
547 0.0 # Default to 0 for successful searches without rate limiting
548 )
549 self._last_results_count = 0
551 def _create_journal_filter(
552 self,
553 engine_name: str,
554 llm: Optional[BaseLLM],
555 settings_snapshot: Optional[Dict[str, Any]],
556 ) -> Optional["JournalReputationFilter"]:
557 """Build the default :class:`JournalReputationFilter` for this engine.
559 Wraps the identical 8-line ``JournalReputationFilter.create_default``
560 boilerplate that previously lived in every academic subclass. The
561 ``engine_name`` is passed explicitly because the auto-derived class
562 name does not match the settings key for ``nasa_ads`` (would be
563 ``"nasaads"``) or ``semantic_scholar`` (would be
564 ``"semanticscholar"``).
566 ``llm`` and ``settings_snapshot`` are passed in rather than read from
567 ``self`` because subclasses build their preview filters *before*
568 calling ``super().__init__()`` (the filter is handed to the parent
569 constructor), so ``self.llm`` / ``self.settings_snapshot`` do not
570 exist yet at call time.
572 Args:
573 engine_name: Settings key identifying the engine (e.g.
574 ``"arxiv"``, ``"nasa_ads"``).
575 llm: Language model used for the filter's relevance pass.
576 settings_snapshot: Settings snapshot for configuration lookups.
578 Returns:
579 A configured :class:`JournalReputationFilter`, or ``None`` if
580 filtering is disabled in settings for this engine.
581 """
582 from ..advanced_search_system.filters.journal_reputation_filter import (
583 JournalReputationFilter,
584 )
586 return JournalReputationFilter.create_default(
587 model=llm, # type: ignore[arg-type]
588 engine_name=engine_name,
589 settings_snapshot=settings_snapshot,
590 )
592 def _configure_programmatic_mode(self, programmatic_mode: bool) -> None:
593 """Set ``programmatic_mode`` and (re)bind the matching rate tracker.
595 Called from ``__init__`` and from the factory as a fallback when an
596 engine subclass swallows the ``programmatic_mode`` kwarg without
597 forwarding it to ``super().__init__``. Safe to call after init —
598 rebinding ``rate_tracker`` discards the previous tracker (no
599 resources to release) and the new one starts with empty in-memory
600 caches.
601 """
602 self.programmatic_mode = programmatic_mode
603 if programmatic_mode:
604 from .rate_limiting.tracker import AdaptiveRateLimitTracker
606 self.rate_tracker = AdaptiveRateLimitTracker(
607 settings_snapshot=self.settings_snapshot,
608 programmatic_mode=programmatic_mode,
609 )
610 else:
611 self.rate_tracker = get_tracker()
613 def get_content_cache(self) -> Any:
614 """Return the effective shared content cache for this engine.
616 Prefers the explicitly configured ``self.content_cache`` (the
617 single cache for the whole research run, threaded through the
618 factory from the strategy), else the cache bound to the current
619 research via the thread-local search context. Returns ``None``
620 when no research cache is active (legacy direct-fetch behaviour).
621 """
622 if getattr(self, "content_cache", None) is not None:
623 return self.content_cache
624 try:
625 from ..utilities.research_content_cache import get_current_cache
626 except Exception: # pragma: no cover - defensive
627 return None
628 try:
629 return get_current_cache()
630 except Exception: # pragma: no cover - defensive
631 return None
633 @staticmethod
634 def _collection_scope_for_item(item: Dict[str, Any]) -> Optional[str]:
635 """Return the normalised collection scope for a library chunk *item*.
637 The same ``(document, chunk_index)`` can hold different text in
638 two collections (different chunk settings), so chunk cache entries
639 are namespaced by collection. Prefers ``collection_id`` (UUID or
640 ``collection_<uuid>``) from metadata/top-level, falling back to a
641 ``collection_*``-prefixed ``collection_name``. Returns ``None``
642 when no collection can be established — callers bypass the shared
643 URL entry rather than risk serving another collection's text.
644 """
645 try:
646 from ..utilities.research_content_cache import (
647 normalize_collection_scope as _norm,
648 )
649 except Exception: # pragma: no cover - defensive
650 return None
651 metadata = item.get("metadata")
652 if not isinstance(metadata, dict): 652 ↛ 653line 652 didn't jump to line 653 because the condition on line 652 was never true
653 metadata = {}
654 for source in (metadata, item):
655 if not isinstance(source, dict): 655 ↛ 656line 655 didn't jump to line 656 because the condition on line 655 was never true
656 continue
657 cid = source.get("collection_id")
658 if isinstance(cid, str) and cid.strip():
659 scoped = _norm(cid)
660 if scoped is not None: 660 ↛ 654line 660 didn't jump to line 654 because the condition on line 660 was always true
661 return scoped
662 for source in (metadata, item):
663 if not isinstance(source, dict): 663 ↛ 664line 663 didn't jump to line 664 because the condition on line 663 was never true
664 continue
665 cname = source.get("collection_name")
666 if isinstance(cname, str) and cname.startswith("collection_"): 666 ↛ 667line 666 didn't jump to line 667 because the condition on line 666 was never true
667 return cname
668 return None
670 @staticmethod
671 def _is_library_chunk_link(link: Any) -> bool:
672 """Return True when *link* is a library chunk URL (``#chunk-N``)."""
673 if not isinstance(link, str): 673 ↛ 674line 673 didn't jump to line 674 because the condition on line 673 was never true
674 return False
675 try:
676 from ..utilities.url_utils import library_display_url
677 except Exception: # pragma: no cover - defensive
678 return False
679 try:
680 display = library_display_url(link.strip())
681 except Exception: # pragma: no cover - defensive
682 return None # type: ignore[return-value]
683 return display is not None and "#chunk-" in display
685 def _get_cache_egress_context(self) -> Any | None:
686 """Egress context for the cache scope gate, memoized per snapshot.
688 Mirrors ``_verify_egress_scope`` memo discipline (identity + policy
689 key) so ADAPTIVE resolution — which can do a bounded DNS lookup —
690 runs once per snapshot, not once per cached URL. Returns None when
691 no snapshot is configured (callers fall back to SSRF-only, matching
692 the network path without a context) or when the policy cannot be
693 evaluated — callers must then refuse the cache hit, matching the
694 network path which disables fetching
695 (``_build_full_search_egress_context`` sets ``include_full_content``
696 to False) instead of fetching SSRF-only.
697 """
698 try:
699 snapshot = getattr(self, "settings_snapshot", None)
700 if not isinstance(snapshot, dict): 700 ↛ 701line 700 didn't jump to line 701 because the condition on line 700 was never true
701 return None
702 strict = _is_strict_scope(snapshot or {})
703 if not snapshot and not strict:
704 return None
705 primary_default = (
706 None if strict else getattr(self, "_engine_name", None)
707 )
708 policy_key = (self._egress_policy_key(), primary_default)
709 cached_ctx = getattr(self, "_cache_egress_ctx", None)
710 cached_key = getattr(self, "_cache_egress_key", None)
711 cached_snap = getattr(self, "_cache_egress_snapshot", None)
712 if (
713 cached_ctx is not None
714 and cached_snap is snapshot
715 and cached_key == policy_key
716 ):
717 return cached_ctx
718 from ..security.egress.policy import (
719 context_from_snapshot,
720 resolve_run_primary_engine,
721 )
722 from ..search_system import username_from_snapshot
724 ctx = context_from_snapshot(
725 snapshot,
726 resolve_run_primary_engine(snapshot, default=primary_default),
727 username=username_from_snapshot(snapshot),
728 )
729 self._cache_egress_ctx = ctx
730 self._cache_egress_key = policy_key
731 self._cache_egress_snapshot = snapshot
732 return ctx
733 except Exception: # pragma: no cover - defensive
734 logger.debug("cache scope gate: egress context unavailable")
735 return None
737 def _is_cache_url_allowed(self, link: Any) -> bool:
738 """Whether a cache hit for *link* may be served to this engine.
740 Defense-in-depth for the shared research cache: content cached by a
741 granted engine (``allow_private_result_fetch=True``) must not flow
742 into a non-granted engine sharing the same run cache. Path-only
743 library routes (``/library/...``, ``/lib/...``) are local DB reads,
744 not network fetches, so they are always allowed. The absolute
745 ``https://library.document/<id>`` alias is NOT exempted — it is a
746 network URL from the engine fill/store path and must pass the same
747 SSRF gate as any web URL. Web URLs must pass the same SSRF gate
748 the fetch path enforces (grant + link-local carve-out).
750 Scope-aware: the run's ``evaluate_url`` scope check must also allow
751 the URL (matching the network path's SSRF + scope pair in
752 ``FullSearchResults.run`` / ``_get_full_content``). Unlike
753 ``FullSearchResults._downloader_allow_private_ips`` — which withholds
754 the private grant from the download pipeline because it cannot
755 enforce scope on redirect hops / browser subresources — the cache
756 has no hops, so the boolean grant is kept and scope is enforced
757 per-URL via ``evaluate_url``. No current producer births a
758 scoped-grant private entry the network path would refuse, so this
759 is latent hardening.
761 Fail-closed twice: when the SSRF validator cannot be imported the
762 hit is refused, and when a settings snapshot is configured but its
763 policy cannot be evaluated the hit is refused too — matching the
764 network path, which disables fetching instead of falling back to
765 SSRF-only. Path-only library routes stay servable even then: they
766 are local DB reads (the fetch tool's ``_try_resolve_url`` fast path
767 likewise skips the egress gate), so refusing them would hide
768 local-only content under a corrupted scope setting. Successful SSRF
769 verdicts are memoized per-run in the shared cache so repeat hits
770 skip DNS (and stay servable when DNS drops mid-run after a
771 success); refusals are recomputed so a transient DNS blip never
772 pins the URL. The scope verdict reuses the run's own DNS memo in
773 ``EgressContext`` and never burns the run-wide denial quota (cache
774 probes use the non-recording evaluation; only real fetches count).
775 That also means cache probes emit no ``policy denied`` audit line
776 while fetch-path denials of the same URL still do — intended
777 (probes must not burn quota), just a triage asymmetry to know
778 about. An exhausted quota still refuses cache probes
779 (``denial_quota_exceeded``), same as the fetch path.
780 """
781 if not isinstance(link, str) or not link.strip(): 781 ↛ 782line 781 didn't jump to line 782 because the condition on line 781 was never true
782 return False
783 # Local-only fast path first: path-only library routes are DB
784 # reads, not network fetches, so they stay servable even when the
785 # run policy is unevaluable. This deliberately mirrors the fetch
786 # tool's strict grammar (32/64-hex or UUID via
787 # ``library_resolver.parse_library_url``), NOT the wider citation
788 # grammar (``is_safe_document_id`` / ``library_display_url`` accepts
789 # any ``[0-9A-Za-z_-]+`` slug): a slug-shaped route can never
790 # resolve via the fetch tool, so exempting it here would let the
791 # two surfaces drift.
792 try:
793 from ..advanced_search_system.tools.fetch.library_resolver import (
794 parse_library_url,
795 )
797 stripped = link.strip()
798 if (
799 stripped.startswith("/")
800 and parse_library_url(stripped) is not None
801 ):
802 return True
803 except Exception: # pragma: no cover - defensive
804 logger.debug("cache grant check: library route parse failed")
805 # Unevaluable policy fails closed for web URLs: a snapshot is
806 # configured but its policy cannot be evaluated (e.g. unknown
807 # egress scope), and the network path disables fetching outright
808 # instead of falling back to SSRF-only.
809 egress_ctx = self._get_cache_egress_context()
810 snapshot = getattr(self, "settings_snapshot", None)
811 if egress_ctx is None and (
812 snapshot or _is_strict_scope(snapshot or {})
813 ):
814 logger.debug(
815 "cache grant check: unevaluable egress policy, refusing hit"
816 )
817 return False
818 try:
819 from ..security.ssrf_validator import validate_url
820 except Exception: # pragma: no cover - defensive
821 logger.debug(
822 "cache grant check: validator unavailable, refusing hit"
823 )
824 return False
825 try:
826 allow_private = bool(
827 getattr(self, "allow_private_result_fetch", False)
828 )
830 def _compute() -> bool:
831 return bool(
832 validate_url(
833 link,
834 allow_private_ips=allow_private,
835 block_link_local=True,
836 )
837 )
839 cache = self.get_content_cache()
840 _memo = getattr(cache, "validated", None)
841 if callable(_memo): 841 ↛ 847line 841 didn't jump to line 847 because the condition on line 841 was always true
842 try:
843 ssrf_ok = bool(_memo(link, allow_private, True, _compute))
844 except Exception: # pragma: no cover - defensive
845 return False
846 else:
847 ssrf_ok = _compute()
848 if not ssrf_ok:
849 return False
850 # Scope axis: the network path (FullSearchResults.run /
851 # _get_full_content) runs evaluate_url after SSRF. The cache
852 # must not serve what that path would deny — but the probe must
853 # not burn the run-wide denial quota either, so it evaluates
854 # without recording; only real fetches count toward
855 # MAX_DENIED_FETCHES_PER_RUN.
856 if egress_ctx is not None:
857 try:
858 from ..security.egress.policy import evaluate_url
860 decision = evaluate_url(
861 link, egress_ctx, record_denial=False
862 )
863 if not decision.allowed:
864 return False
865 except Exception: # pragma: no cover - defensive
866 return False
867 return True
868 except Exception: # pragma: no cover - defensive
869 return False
871 def _fill_full_content_from_cache(
872 self, items: list[Dict[str, Any]]
873 ) -> tuple[list[Dict[str, Any]], list[Dict[str, Any]]]:
874 """Split *items* into (already_cached, needs_fetch).
876 Cached items get ``full_content`` filled in place from the shared
877 research cache. Items without a link, or with no cache entry, go
878 to *needs_fetch*. Returns ``([], items)`` when no cache is active.
880 Library chunk URLs are namespaced by collection: the same
881 document/chunk URL in two collections holds different text, so a
882 chunk whose collection cannot be established bypasses the shared
883 entry (fresh DB lookup) rather than risking another collection's
884 text. Document-level library URLs (``Document.text_content`` is
885 collection-independent) and web URLs use the URL-only key.
886 """
887 cache = self.get_content_cache()
888 if cache is None or not items:
889 return [], list(items)
890 cached: list[Dict[str, Any]] = []
891 missing: list[Dict[str, Any]] = []
892 for item in items:
893 if not isinstance(item, dict): 893 ↛ 894line 893 didn't jump to line 894 because the condition on line 893 was never true
894 missing.append(item)
895 continue
896 link = item.get("link") or item.get("url")
897 if not link: 897 ↛ 898line 897 didn't jump to line 898 because the condition on line 897 was never true
898 missing.append(item)
899 continue
900 if not self._is_cache_url_allowed(link):
901 missing.append(item)
902 continue
903 try:
904 if self._is_library_chunk_link(link):
905 scope = self._collection_scope_for_item(item)
906 if scope is None:
907 missing.append(item)
908 continue
909 hit = cache.get(link, collection_id=scope)
910 else:
911 hit = cache.get(link)
912 except Exception: # pragma: no cover - defensive
913 hit = None
914 if hit is not None and isinstance(hit.get("content"), str):
915 # A preview can already carry an abstract in full_content.
916 # The shared cache contains complete text and takes priority.
917 if (
918 not item.get("full_content")
919 or item.get("full_content_is_complete") is not True
920 ):
921 item["full_content"] = hit["content"]
922 item["full_content_is_complete"] = True
923 cached.append(item)
924 else:
925 missing.append(item)
926 return cached, missing
928 def _store_full_content_in_cache(self, items: list[Dict[str, Any]]) -> None:
929 """Store only explicitly complete text in the shared research cache.
931 Engines must set ``full_content_is_complete=True`` at the point
932 they retrieve source text. The legacy ``full_content`` field also
933 holds abstracts, snippets, errors and retrieved fragments; those
934 must not satisfy a later full-document fetch or prevent a retry.
935 Unknown producers remain usable as search results without caching.
937 Library chunk entries are namespaced by collection (see
938 :meth:`_fill_full_content_from_cache`); chunks without an
939 establishable collection are not stored under the shared URL key.
940 Entries this engine's grant would not allow to read are not
941 written either, keeping the shared cache grant-consistent.
942 """
943 cache = self.get_content_cache()
944 if cache is None or not items:
945 return
946 for item in items:
947 if not isinstance(item, dict): 947 ↛ 948line 947 didn't jump to line 948 because the condition on line 947 was never true
948 continue
949 if item.get("full_content_is_complete") is not True:
950 continue
951 link = item.get("link") or item.get("url")
952 content = item.get("full_content")
953 if not link or not isinstance(content, str) or not content: 953 ↛ 954line 953 didn't jump to line 954 because the condition on line 953 was never true
954 continue
955 if not self._is_cache_url_allowed(link): 955 ↛ 956line 955 didn't jump to line 956 because the condition on line 955 was never true
956 continue
957 try:
958 if self._is_library_chunk_link(link):
959 scope = self._collection_scope_for_item(item)
960 if scope is None:
961 continue
962 cache.put(
963 link,
964 content,
965 title=item.get("title"),
966 url_out=link,
967 collection_id=scope,
968 )
969 else:
970 cache.put(
971 link,
972 content,
973 title=item.get("title"),
974 url_out=link,
975 )
976 except Exception: # pragma: no cover - defensive
977 continue
979 def _get_full_content_cached(
980 self, relevant_items: list[Dict[str, Any]]
981 ) -> list[Dict[str, Any]]:
982 """Cache-aware wrapper around ``_get_full_content``.
984 Serves repeat URLs (quick phase vs subagents, lead vs subagents,
985 repeated ``web_search`` calls) from the single research-scoped
986 cache and only fetches uncached URLs on the spot. Newly fetched
987 content is stored for later consumers in the same research run.
988 """
989 if not relevant_items:
990 return []
991 _, needs_fetch = self._fill_full_content_from_cache(relevant_items)
992 if not needs_fetch:
993 return relevant_items
994 # Fetch only the uncached subset; cached items were filled in place.
995 fetched = self._get_full_content(needs_fetch)
996 # _get_full_content may return a new list; merge back by identity:
997 # cached items are already correct in relevant_items, fetched ones
998 # carry fresh full_content to store.
999 try:
1000 self._store_full_content_in_cache(
1001 fetched if isinstance(fetched, list) else needs_fetch
1002 )
1003 except Exception: # allow: silent-exception
1004 pass
1005 if isinstance(fetched, list) and fetched is not needs_fetch:
1006 # Engine returned a new list for the subset: splice it back in
1007 # order. Fall back to the in-place list on any shape mismatch.
1008 # The contract is length-preserving and order-preserving —
1009 # ``zip`` truncation would silently drop results if an engine
1010 # returned fewer/reordered items, so pin it with a warning.
1011 if len(fetched) != len(needs_fetch): 1011 ↛ 1012line 1011 didn't jump to line 1012 because the condition on line 1011 was never true
1012 logger.warning(
1013 "_get_full_content returned {} items for {} "
1014 "requested; falling back to in-place list",
1015 len(fetched),
1016 len(needs_fetch),
1017 )
1018 return relevant_items
1019 try:
1020 by_id = {id(o): n for o, n in zip(needs_fetch, fetched)}
1021 merged: list[Dict[str, Any]] = []
1022 for item in relevant_items:
1023 merged.append(by_id.get(id(item), item))
1024 return merged
1025 except Exception: # pragma: no cover - defensive
1026 return relevant_items
1027 return relevant_items
1029 @property
1030 def max_filtered_results(self) -> int:
1031 """Get the maximum number of filtered results."""
1032 return self._max_filtered_results
1034 @max_filtered_results.setter
1035 def max_filtered_results(self, value: int) -> None:
1036 """Set the maximum number of filtered results."""
1037 if value is None:
1038 value = DEFAULT_MAX_FILTERED_RESULTS
1039 logger.warning(
1040 f"Setting max_filtered_results to {DEFAULT_MAX_FILTERED_RESULTS}"
1041 )
1042 self._max_filtered_results = int(value)
1044 @property
1045 def max_results(self) -> int:
1046 """Get the maximum number of search results."""
1047 return self._max_results
1049 @max_results.setter
1050 def max_results(self, value: int) -> None:
1051 """Set the maximum number of search results."""
1052 if value is None:
1053 value = 10
1054 self._max_results = max(1, int(value))
1056 def _get_adaptive_wait(self) -> float:
1057 """Get adaptive wait time from tracker."""
1058 wait_time = self.rate_tracker.get_wait_time(self.engine_type)
1059 self._last_wait_time = wait_time
1060 logger.debug(
1061 f"{self.engine_type} waiting {wait_time:.2f}s before retry"
1062 )
1063 return wait_time
1065 def _record_retry_outcome(self, retry_state) -> None:
1066 """Record outcome after retry completes."""
1067 success = (
1068 not retry_state.outcome.failed if retry_state.outcome else False
1069 )
1070 self.rate_tracker.record_outcome(
1071 self.engine_type,
1072 self._last_wait_time or 0,
1073 success,
1074 retry_state.attempt_number,
1075 error_type="RateLimitError" if not success else None,
1076 search_result_count=self._last_results_count if success else 0,
1077 )
1079 def _verify_egress_scope(self) -> None:
1080 """Runtime backstop: verify this engine is allowed under the egress
1081 scope in ``self.settings_snapshot`` before executing a search.
1083 No-op when ``settings_snapshot`` is missing or empty (programmatic
1084 API callers) unless the operator enforces STRICT, or when
1085 ``_engine_name`` has not been set. Raises
1086 ``PolicyDeniedError`` when the policy denies the engine; any other
1087 internal error in the policy evaluation is logged and ignored so a
1088 broken backstop never takes down searches the factory PEP already
1089 approved.
1091 This is a defense-in-depth check behind the factory PEP, the
1092 strategy-level filters, and the audit hook. It covers engines that
1093 bypassed the factory (direct instantiation with a snapshot).
1094 Changes to the captured scope or primary are caught both when the
1095 caller refreshes ``settings_snapshot`` and when it edits those
1096 keys in place.
1098 The verification is memoized per snapshot IDENTITY plus the
1099 policy-relevant VALUES (scope, primary): the full policy inputs
1100 are stable for a given snapshot object, so re-evaluating on every
1101 ``run()`` would only repeat the same decision — and under
1102 ADAPTIVE scope with a URL-configurable primary the evaluation can
1103 include a DNS lookup (bounded at 2s), which must not become a
1104 per-search stall. Assigning a refreshed snapshot OR mutating the
1105 scope/primary keys in place invalidates the memo and re-verifies.
1106 Denials are never memoized (they raise).
1107 """
1108 if not self.settings_snapshot and not _is_strict_scope(None):
1109 return
1110 if not self._engine_name:
1111 # A snapshot WITHOUT a stamped engine name means a direct
1112 # instantiation the factory never saw and no subclass
1113 # self-stamps — the backstop cannot evaluate it. Surface the
1114 # gap in the audit log (once per instance) instead of
1115 # skipping silently.
1116 if not self._egress_skip_warned:
1117 self._egress_skip_warned = True
1118 logger.bind(policy_audit=True).warning(
1119 "egress backstop skipped: engine has a settings "
1120 "snapshot but no _engine_name stamped",
1121 engine_type=self.__class__.__name__,
1122 )
1123 return
1124 if (
1125 self._egress_verified_snapshot is not None
1126 and self.settings_snapshot is self._egress_verified_snapshot
1127 and self._egress_policy_key() == self._egress_verified_policy_key
1128 ):
1129 return
1130 from ..security.egress.policy import PolicyDeniedError
1132 try:
1133 self._check_egress_policy()
1134 except PolicyDeniedError:
1135 logger.bind(policy_audit=True).warning(
1136 "runtime egress backstop denied engine",
1137 engine=self._engine_name,
1138 )
1139 raise
1140 except Exception:
1141 logger.bind(policy_audit=True).debug(
1142 "egress runtime verify errored; "
1143 "factory PEP still enforces at instantiation",
1144 engine=self._engine_name,
1145 )
1146 else:
1147 self._egress_verified_snapshot = self.settings_snapshot
1148 self._egress_verified_policy_key = self._egress_policy_key()
1150 def _egress_policy_key(self) -> tuple:
1151 """The policy-relevant snapshot values the backstop memo guards on.
1153 Cheap dict reads only — no context construction, no DNS. Not an
1154 exhaustive input set (URL settings or DB-side collection flags can
1155 also influence the decision), but it covers the keys an in-place
1156 mutation would realistically target; the factory PEP remains the
1157 primary enforcement for everything else.
1159 MAINTENANCE: if a new snapshot key ever becomes policy-relevant in
1160 ``context_from_snapshot``/``evaluate_engine``, add it here too —
1161 otherwise an in-place mutation of that key returns a stale memo.
1163 """
1164 scope = unwrap_setting(
1165 self.settings_snapshot.get("policy.egress_scope")
1166 )
1167 primary = unwrap_setting(self.settings_snapshot.get("search.tool"))
1168 return (scope, primary)
1170 def _check_egress_policy(self) -> None:
1171 """Inner helper so the raise is not inside the except handler
1172 (ruff TRY301). Raises PolicyDeniedError on scope mismatch."""
1173 from ..security.egress.policy import (
1174 Decision,
1175 PolicyDeniedError,
1176 context_from_snapshot,
1177 evaluate_engine,
1178 resolve_run_primary_engine,
1179 )
1180 from ..search_system import username_from_snapshot
1182 try:
1183 # Match the factory: STRICT requires the saved primary and must
1184 # never let this engine approve itself as a fallback.
1185 primary_default = (
1186 None
1187 if _is_strict_scope(self.settings_snapshot)
1188 else self._engine_name
1189 )
1190 ctx = context_from_snapshot(
1191 self.settings_snapshot,
1192 resolve_run_primary_engine(
1193 self.settings_snapshot, default=primary_default
1194 ),
1195 username=username_from_snapshot(self.settings_snapshot),
1196 )
1197 except ValueError as exc:
1198 # The backstop propagates policy denials but ignores other
1199 # errors; invalid configuration must stop a cached engine too.
1200 raise PolicyDeniedError(
1201 Decision(False, "invalid_policy_config"),
1202 target=self._engine_name,
1203 ) from exc
1204 decision = evaluate_engine(
1205 self._engine_name,
1206 ctx,
1207 settings_snapshot=self.settings_snapshot,
1208 )
1209 if not decision.allowed:
1210 raise PolicyDeniedError(decision, target=self._engine_name)
1212 def run(
1213 self, query: str, research_context: Dict[str, Any] | None = None
1214 ) -> List[Dict[str, Any]]:
1215 """
1216 Run the search engine with a given query, retrieving and filtering results.
1217 This implements a two-phase retrieval approach:
1218 1. Get preview information for many results
1219 2. Filter the previews for relevance
1220 3. Get full content for only the relevant results
1222 Args:
1223 query: The search query
1224 research_context: Context from previous research to use.
1226 Returns:
1227 List of search results with full content (if available)
1228 """
1229 state_var = self._search_run_state_var()
1230 token = state_var.set(self._new_search_run_state())
1231 try:
1232 return self._run_search(query, research_context)
1233 finally:
1234 # Reset also restores the outer state for a nested run(), and
1235 # cannot clear another thread's in-flight cache or outcome flags.
1236 state_var.reset(token)
1238 def _run_search(
1239 self, query: str, research_context: Dict[str, Any] | None = None
1240 ) -> List[Dict[str, Any]]:
1241 """Execute and record one search inside its invocation state."""
1242 logger.info(f"---Execute a search using {self.__class__.__name__}---")
1244 # Runtime egress scope backstop: verify this engine is allowed
1245 # before making any HTTP requests.
1246 self._verify_egress_scope()
1248 # Track search call for metrics (if available and not in programmatic mode)
1249 should_record_metrics = False
1250 context_was_set = False
1251 if not self.programmatic_mode:
1252 from ..metrics.search_tracker import SearchTracker
1254 should_record_metrics = True
1256 # For thread-safe context propagation: if we have research_context parameter, use it
1257 # Otherwise, try to inherit from current thread context (normal case)
1258 # This allows strategies running in threads to explicitly pass context when needed
1259 if research_context:
1260 # Explicit context provided - use it and set it for this thread
1261 set_search_context(research_context)
1262 context_was_set = True
1264 engine_name = metrics_engine_name(self)
1265 start_time = time.time()
1267 success = True
1268 error_message = None
1269 error_type = None
1270 results_count = 0
1272 # Define the core search function with retry logic. Each branch
1273 # defines its own function and both are selected through
1274 # ``run_search``: conditionally redefining a single name is a
1275 # mypy error since tenacity 9.2, whose @retry returns a typed
1276 # wrapper (_RetryDecorated) instead of a plain Callable.
1277 run_search: Callable[[], Any]
1278 if self.rate_tracker.enabled:
1279 # Rate limiting enabled - use retry with adaptive wait
1280 @retry(
1281 stop=stop_after_attempt(3),
1282 wait=AdaptiveWait(lambda: self._get_adaptive_wait()),
1283 retry=retry_if_exception_type((RateLimitError,)),
1284 after=self._record_retry_outcome,
1285 reraise=True,
1286 )
1287 def _run_with_retry():
1288 nonlocal success, error_message, results_count
1289 return _execute_search()
1291 run_search = _run_with_retry
1292 else:
1293 # Rate limiting disabled - run without retry
1294 def _run_without_retry():
1295 nonlocal success, error_message, results_count
1296 return _execute_search()
1298 run_search = _run_without_retry
1300 def _execute_search():
1301 nonlocal success, error_message, error_type, results_count
1303 previews_received = False
1304 try:
1305 # Step 1: Get preview information for items
1306 self._search_failed = False
1307 self._rate_limited_after_previews = False
1308 self.last_search_failure = None
1309 previews = self._get_previews(query)
1310 if self._search_failed:
1311 success = False
1312 error_message = "Preview retrieval failed"
1313 if not previews:
1314 logger.info(
1315 f"Search engine {self.__class__.__name__} returned no preview results"
1316 )
1317 results_count = 0
1318 return []
1319 # The search request worked. A partial failure while
1320 # gathering previews, or the relevance filter dropping every
1321 # result below, does not make the engine unavailable.
1322 previews_received = True
1323 self.last_search_failure = None
1325 # Pre-filter enrichment: resolve DOIs to OpenAlex source
1326 # IDs BEFORE the preview filters run. The
1327 # JournalReputationFilter is registered as a preview
1328 # filter and uses ``result["openalex_source_id"]`` for
1329 # Tier 2 lookups; running enrichment afterwards (as the
1330 # old pipeline did) left the field empty at filter time,
1331 # silently forcing a fragile-name-match fallback. Only
1332 # for scientific engines whose results carry DOIs.
1333 if getattr(self, "is_scientific", False):
1334 try:
1335 from ..utilities.openalex_enrichment import (
1336 enrich_results_with_source_ids,
1337 )
1339 email = getattr(self, "email", None)
1340 previews = enrich_results_with_source_ids(
1341 previews,
1342 email=email,
1343 api_key=self._resolve_openalex_enrichment_key(),
1344 # Enrichment's own drop is call-local; this
1345 # latches it on the engine so later runs
1346 # resolve None instead of the same refused
1347 # key, and keeps the literal redactable.
1348 on_key_rejected=self._note_openalex_key_rejected,
1349 )
1350 except Exception:
1351 logger.debug(
1352 "DOI enrichment skipped (import or call failed)"
1353 )
1355 for preview_filter in self._preview_filters:
1356 previews = preview_filter.filter_results(previews, query)
1358 # Step 2: Filter previews for relevance with LLM
1359 enable_llm_filter = getattr(
1360 self, "enable_llm_relevance_filter", False
1361 )
1363 if enable_llm_filter and self.llm:
1364 filtered_items = self._filter_for_relevance(previews, query)
1365 else:
1366 filtered_items = previews
1367 logger.debug(
1368 f"[{type(self).__name__}] Relevance filter skipped "
1369 f"(enabled={enable_llm_filter}, "
1370 f"llm={'yes' if self.llm else 'no'})"
1371 )
1373 # Step 3: Get full content for filtered items
1374 if self.search_snippets_only:
1375 logger.info("Returning snippet-only results as per config")
1376 results = filtered_items
1377 else:
1378 # Cache-aware: repeat URLs across the quick phase,
1379 # subsections and subagents are served from the single
1380 # research-scoped cache instead of re-fetched on the spot.
1381 results = self._get_full_content_cached(filtered_items)
1383 for content_filter in self._content_filters:
1384 results = content_filter.filter_results(results, query)
1386 results_count = len(results)
1387 self._last_results_count = results_count
1389 # Record success if we get here and rate limiting is enabled
1390 if (
1391 self.rate_tracker.enabled
1392 and self._rate_limited_after_previews
1393 ):
1394 # The results are usable, but the provider answered a
1395 # later request with HTTP 429: tell the limiter, as
1396 # ``_record_retry_outcome`` does for a raised 429.
1397 logger.info(
1398 f"Recording rate-limited search for {self.engine_type}: wait_time={self._last_wait_time}s, results={results_count}"
1399 )
1400 self.rate_tracker.record_outcome(
1401 self.engine_type,
1402 self._last_wait_time or 0,
1403 success=False,
1404 retry_count=1,
1405 error_type="RateLimitError",
1406 search_result_count=results_count,
1407 )
1408 elif self.rate_tracker.enabled:
1409 logger.info(
1410 f"Recording successful search for {self.engine_type}: wait_time={self._last_wait_time}s, results={results_count}"
1411 )
1412 self.rate_tracker.record_outcome(
1413 self.engine_type,
1414 self._last_wait_time,
1415 success=True,
1416 retry_count=1, # First attempt succeeded
1417 search_result_count=results_count,
1418 )
1419 else:
1420 logger.info(
1421 f"Rate limiting disabled, not recording search for {self.engine_type}"
1422 )
1424 return results
1426 except RateLimitError as exc:
1427 # Only re-raise if rate limiting is enabled
1428 if self.rate_tracker.enabled:
1429 raise
1430 self._record_search_failure(exc)
1431 # If rate limiting is disabled, treat as regular error
1432 success = False
1433 error_message = "Rate limit hit but rate limiting disabled"
1434 error_type = type(exc).__name__
1435 logger.warning(
1436 f"Rate limit hit on {self.__class__.__name__} but rate limiting is disabled"
1437 )
1438 results_count = 0
1439 return []
1440 except Exception as e:
1441 if not previews_received:
1442 # Past the previews, the provider already answered: a
1443 # filter or content error is not an engine outage.
1444 self._record_search_failure(e)
1445 # Other errors - don't retry
1446 success = False
1447 # Sanitize before it flows to SearchTracker.record_search
1448 # (and the database) and to the log. Use logger.warning with
1449 # the scrubbed text instead of logger.exception: the
1450 # cause chain frequently carries the request URL or auth
1451 # header from upstream HTTP clients (see #4131).
1452 error_message = safe_msg = self._scrub_error(e)
1453 error_type = type(e).__name__
1454 logger.warning(
1455 f"Search engine {self.__class__.__name__} failed: {safe_msg}"
1456 )
1457 results_count = 0
1458 return []
1460 try:
1461 return run_search() # type: ignore[no-any-return]
1462 except Exception as e:
1463 # Includes the last RateLimitError once retries are exhausted:
1464 # the decorator re-raises it (reraise=True), never RetryError.
1465 self._record_search_failure(e)
1466 success = False
1467 error_message = safe_msg = self._scrub_error(e)
1468 error_type = type(e).__name__
1469 logger.warning(
1470 f"Search engine {self.__class__.__name__} error: {safe_msg}"
1471 )
1472 return []
1473 finally:
1474 try:
1475 # Record search metrics BEFORE clearing context (record_search needs it)
1476 if should_record_metrics:
1477 response_time_ms = int((time.time() - start_time) * 1000)
1478 SearchTracker.record_search(
1479 engine_name=engine_name,
1480 query=query,
1481 results_count=results_count,
1482 response_time_ms=response_time_ms,
1483 success=success,
1484 error_message=error_message,
1485 error_type=error_type,
1486 )
1487 finally:
1488 # Clean up temporary search result storage
1489 for attr in self._temp_attributes():
1490 if hasattr(self, attr):
1491 delattr(self, attr)
1492 # ALWAYS clean up search context, even if recording fails
1493 if context_was_set:
1494 clear_search_context()
1496 def invoke(self, query: str) -> List[Dict[str, Any]]:
1497 """Compatibility method for LangChain tools"""
1498 return self.run(query)
1500 def _filter_for_relevance(
1501 self, previews: List[Dict[str, Any]], query: str
1502 ) -> List[Dict[str, Any]]:
1503 """
1504 Filter search results by relevance using the LLM.
1506 Delegates to the ``relevance_filter`` module, which prompts the
1507 LLM for a plain-text list of relevant indices and parses them
1508 with a regex (no structured output).
1510 Args:
1511 previews: List of preview dictionaries
1512 query: The original search query
1514 Returns:
1515 Filtered list of preview dictionaries
1516 """
1517 engine_name = self._engine_name or type(self).__name__
1519 if not self.llm or len(previews) <= 1:
1520 logger.debug(
1521 f"[{engine_name}] Skipping relevance filter "
1522 f"(llm={'yes' if self.llm else 'no'}, "
1523 f"previews={len(previews)})"
1524 )
1525 return previews
1527 from .relevance_filter import filter_previews_for_relevance
1529 return filter_previews_for_relevance(
1530 llm=self.llm,
1531 previews=previews,
1532 query=query,
1533 max_filtered_results=self.max_filtered_results,
1534 engine_name=engine_name,
1535 batch_size=self.relevance_filter_batch_size,
1536 max_parallel_batches=self.relevance_filter_max_parallel_batches,
1537 )
1539 # =========================================================================
1540 # Shared Helper Methods for Subclasses
1541 # =========================================================================
1543 def _resolve_openalex_enrichment_key(self) -> Optional[str]:
1544 """Resolve the OpenAlex API key used by the DOI enrichment pass.
1546 The enrichment request always goes to ``api.openalex.org``, no
1547 matter which engine produced the previews, so every
1548 ``is_scientific`` engine (arxiv, semantic_scholar, zenodo,
1549 nasa_ads, pubchem, pubmed, …) needs the *OpenAlex* credential
1550 here — not its own. Only ``OpenAlexSearchEngine`` sets the
1551 ``openalex_api_key`` attribute, so the settings snapshot is the
1552 source for everyone else.
1554 Deliberately never reads ``self.api_key``: sending, say, a NASA
1555 ADS key to OpenAlex would hand one vendor's credential to another
1556 host. The isolation is pinned by
1557 ``test_openalex_enrichment_key_isolated_from_other_engine_keys``.
1558 """
1559 from ..utilities.openalex_enrichment import (
1560 normalize_openalex_api_key,
1561 )
1563 if self._openalex_key_rejected:
1564 # OpenAlex already refused this installation's key on this
1565 # engine. Clearing ``self.openalex_api_key`` alone would not
1566 # help: the snapshot fallback below is where the key came
1567 # from in the first place, so without this latch the
1568 # enrichment request of the very same ``run()`` re-sends it.
1569 return None
1571 key = normalize_openalex_api_key(
1572 getattr(self, "openalex_api_key", None)
1573 )
1574 if key:
1575 return key
1577 settings_snapshot = getattr(self, "settings_snapshot", None)
1578 if not settings_snapshot:
1579 return None
1581 # An explicit default is required: with no ``default`` argument
1582 # at all, a non-empty snapshot that lacks this key and no
1583 # thread-local settings context makes ``get_setting_from_snapshot``
1584 # raise ``NoSettingsContextError``. That would propagate out of
1585 # this resolver in the *argument* position of the enrichment call
1586 # and be swallowed by the caller's blanket handler, silently
1587 # skipping DOI enrichment for every scientific engine. The empty
1588 # string (rather than None) is the safer spelling because it is
1589 # falsy under every version of that function's "was a default
1590 # supplied?" test, and ``normalize_openalex_api_key`` maps it to
1591 # None, i.e. "enrich keylessly".
1592 return normalize_openalex_api_key(
1593 get_setting_from_snapshot(
1594 "search.engine.web.openalex.api_key",
1595 default="",
1596 settings_snapshot=settings_snapshot,
1597 )
1598 )
1600 @staticmethod
1601 def _is_valid_api_key(api_key: Optional[str]) -> bool:
1602 """
1603 Check if an API key is valid (not a placeholder value).
1605 Args:
1606 api_key: The API key to validate
1608 Returns:
1609 True if the key appears to be a real API key, False if it's a placeholder
1611 Example:
1612 >>> BaseSearchEngine._is_valid_api_key("sk-abc123")
1613 True
1614 >>> BaseSearchEngine._is_valid_api_key("YOUR_API_KEY_HERE")
1615 False
1616 """
1617 return not _is_api_key_placeholder(api_key)
1619 @staticmethod
1620 def _extract_display_link(url: str, fallback: str = "") -> str:
1621 """Extract the netloc (domain) from a URL for display purposes.
1623 Args:
1624 url: The URL to extract from
1625 fallback: Value to return on parse failure (default: empty string)
1627 Returns:
1628 The netloc portion of the URL, or *fallback* if parsing fails.
1629 """
1630 if not url:
1631 return fallback
1632 try:
1633 parsed = urlparse(url)
1634 return parsed.netloc or fallback
1635 except Exception:
1636 return fallback
1638 @staticmethod
1639 def _clean_result_url(value: Any) -> str:
1640 """Normalize a raw search-result URL for the validity gate.
1642 Coerces ``None`` (and any other falsy value) to ``""`` and any
1643 truthy value to ``str`` before stripping surrounding whitespace.
1645 The strip is **load-bearing**, not merely cosmetic: several engines
1646 gate the URL on a ``url.lower().startswith(("http://", "https://"))``
1647 prefix check (see ``_is_valid_search_result``) *before* it reaches
1648 the SSRF validator's own internal strip, so leading/trailing
1649 whitespace — common in HTML-scraped ``href`` attributes — would
1650 silently drop an otherwise-valid result. Cleaning once at extraction
1651 also keeps the URL tidy for logs and downstream consumers (preview
1652 ``id``/``link`` fields, etc.).
1654 Args:
1655 value: The raw URL value from a search result, e.g.
1656 ``result.get("url")`` or a parsed HTML ``href`` attribute.
1658 Returns:
1659 The whitespace-stripped URL string, or ``""`` if *value* is
1660 falsy (``None``, empty, etc.).
1661 """
1662 if not value:
1663 return ""
1664 return str(value).strip()
1666 def _resolve_api_key(
1667 self,
1668 api_key: Optional[str],
1669 setting_key: str,
1670 engine_name: str = "search engine",
1671 settings_snapshot: Optional[Dict[str, Any]] = None,
1672 ) -> str:
1673 """
1674 Resolve an API key from multiple sources with priority order.
1676 Environment variables are handled automatically by SettingsManager
1677 when building the settings snapshot, so they don't need to be
1678 checked separately here.
1680 Priority order:
1681 1. Direct parameter (api_key argument)
1682 2. Settings snapshot (via setting_key)
1684 Args:
1685 api_key: API key passed directly as parameter
1686 setting_key: Key to look up in settings snapshot (e.g., "search.brave_api_key")
1687 engine_name: Human-readable engine name for error messages
1688 settings_snapshot: Optional settings snapshot dict (uses self.settings_snapshot if not provided)
1690 Returns:
1691 The resolved API key string
1693 Raises:
1694 ValueError: If no valid API key is found from any source
1696 Example:
1697 >>> engine._resolve_api_key(
1698 ... api_key=None,
1699 ... setting_key="search.brave_api_key",
1700 ... engine_name="Brave Search"
1701 ... )
1702 "sk-abc123..."
1703 """
1704 # Use instance settings snapshot if not provided
1705 if settings_snapshot is None:
1706 settings_snapshot = self.settings_snapshot
1708 # Priority 1: Direct parameter
1709 if self._is_valid_api_key(api_key) and api_key is not None:
1710 return api_key.strip()
1712 # Priority 2: Settings snapshot (includes env var overrides via SettingsManager)
1713 if settings_snapshot:
1714 settings_value = get_setting_from_snapshot(
1715 setting_key,
1716 default=None,
1717 settings_snapshot=settings_snapshot,
1718 )
1719 if self._is_valid_api_key(settings_value):
1720 return settings_value.strip() if settings_value else ""
1722 # No valid API key found
1723 masked_key = self._mask_api_key(str(api_key)) if api_key else "None"
1724 raise ValueError(
1725 f"No valid API key found for {engine_name}. "
1726 f"Checked: direct parameter ({masked_key}), "
1727 f"settings key '{setting_key}'. "
1728 f"Please provide a valid API key."
1729 )
1731 def _record_search_failure(self, error: BaseException) -> None:
1732 """Preserve request failures even when an engine returns an empty list.
1734 Call at search request catch sites, not optional enrichment/fetch paths.
1735 A later successful attempt or usable results supersede this signal.
1736 An error specific to this query (HTTP 400) is not recorded: the
1737 engine itself still works.
1738 """
1739 if is_query_error(error):
1740 return
1741 bare_403_is_bad_key = self.bare_403_means_bad_api_key and bool(
1742 getattr(self, "api_key", None)
1743 )
1744 self.last_search_failure = (
1745 failure_from_exception(
1746 error, bare_403_is_bad_key=bare_403_is_bad_key
1747 )
1748 or self.last_search_failure
1749 or SearchFailure(
1750 "Search request failed; a later attempt may succeed"
1751 )
1752 )
1754 def _is_rate_limit_error(
1755 self,
1756 error: Union[Exception, str, int],
1757 additional_patterns: Optional[Set[str]] = None,
1758 ) -> bool:
1759 """
1760 Detect if an error is a rate limit error.
1762 Checks multiple sources for rate limit indicators:
1763 - HTTP status code 429
1764 - HTTPError response objects
1765 - Error messages containing rate limit phrases
1767 Args:
1768 error: The error to check (Exception, string, or HTTP status code)
1769 additional_patterns: Optional set of additional patterns to match
1771 Returns:
1772 True if the error appears to be a rate limit error
1774 Example:
1775 >>> engine._is_rate_limit_error(429)
1776 True
1777 >>> engine._is_rate_limit_error("Rate limit exceeded")
1778 True
1779 >>> engine._is_rate_limit_error(ValueError("Invalid input"))
1780 False
1781 """
1782 # Combine default and additional patterns
1783 patterns = self.rate_limit_patterns.copy()
1784 if additional_patterns:
1785 patterns.update(additional_patterns)
1787 # Check integer status code directly
1788 if isinstance(error, int):
1789 return error == 429
1791 # Convert to string for pattern matching
1792 error_str = ""
1793 status_code = None
1795 if isinstance(error, str):
1796 error_str = error
1797 elif isinstance(error, Exception):
1798 error_str = str(error)
1800 # Check for HTTP status code in common HTTP error types
1801 if hasattr(error, "status_code"):
1802 status_code = error.status_code
1803 elif hasattr(error, "response"):
1804 response = error.response
1805 if hasattr(response, "status_code"):
1806 status_code = response.status_code
1808 # Check status code first
1809 if status_code == 429:
1810 return True
1812 # Case-insensitive pattern matching
1813 error_lower = error_str.lower()
1814 for pattern in patterns:
1815 if pattern.lower() in error_lower:
1816 return True
1818 return False
1820 def _raise_if_rate_limit(
1821 self,
1822 error: Union[Exception, str, int],
1823 additional_patterns: Optional[Set[str]] = None,
1824 ) -> None:
1825 """
1826 Raise RateLimitError if the given error is a rate limit error.
1828 Convenience method that combines _is_rate_limit_error check with
1829 raising RateLimitError.
1831 Args:
1832 error: The error to check
1833 additional_patterns: Optional set of additional patterns to match
1835 Raises:
1836 RateLimitError: If the error is detected as a rate limit error
1838 Example:
1839 >>> try:
1840 ... response = make_api_call()
1841 ... except Exception as e:
1842 ... engine._raise_if_rate_limit(e)
1843 """
1844 if self._is_rate_limit_error(error, additional_patterns):
1845 error_msg = str(error) if not isinstance(error, str) else error
1846 raise RateLimitError(
1847 f"Rate limit detected: {self._sanitize_error_message(error_msg)}",
1848 retry_after=retry_after_from_error(error),
1849 )
1851 def _extract_full_result(self, item: Dict[str, Any]) -> Dict[str, Any]:
1852 """
1853 Extract the full result from an item that may contain a _full_result key.
1855 This is a helper for the default _get_full_content implementation.
1856 It extracts data from the _full_result key if present, otherwise uses
1857 the item directly, and removes the internal _full_result key.
1859 Args:
1860 item: A search result item that may contain a _full_result key
1862 Returns:
1863 A dictionary with the full result data, without the _full_result key
1865 Example:
1866 >>> engine._extract_full_result({"title": "A", "_full_result": {"title": "A", "content": "Full"}})
1867 {"title": "A", "content": "Full"}
1868 """
1869 source = item.get("_full_result")
1870 if source is None:
1871 source = item
1872 return {k: v for k, v in source.items() if k != "_full_result"}
1874 def _get_full_content(
1875 self, relevant_items: List[Dict[str, Any]]
1876 ) -> List[Dict[str, Any]]:
1877 """
1878 Get full content for the relevant items.
1880 Default implementation extracts data from _full_result keys if present.
1881 Subclasses can override this method to fetch additional content from
1882 external sources (e.g., web scraping, API calls).
1884 Args:
1885 relevant_items: List of relevant preview dictionaries
1887 Returns:
1888 List of result dictionaries with full content
1890 Example:
1891 >>> engine._get_full_content([
1892 ... {"title": "A", "_full_result": {"title": "A", "content": "Full A"}},
1893 ... {"title": "B"}
1894 ... ])
1895 [{"title": "A", "content": "Full A"}, {"title": "B"}]
1896 """
1897 if not relevant_items:
1898 return []
1899 return [self._extract_full_result(item) for item in relevant_items]
1901 def _build_full_search_egress_context(self):
1902 """Build an EgressContext from self.settings_snapshot for the
1903 per-URL scope gate inside FullSearchResults. Returns None when a
1904 context can't be built and disables full-content fetching when
1905 policy evaluation fails on a supplied snapshot.
1907 The previous "fail closed by returning None" comment was wrong:
1908 full_search.py treats ``egress_context is None`` as
1909 ``evaluate_url_fn = None`` and skips the per-URL scope check
1910 entirely, leaving only SSRF — which allows public hosts under
1911 PRIVATE_ONLY. When the snapshot was supplied but the policy is
1912 unevaluable, we now turn full-content fetching off for this
1913 engine instead.
1914 """
1915 snapshot = getattr(self, "settings_snapshot", None)
1916 if not snapshot and not _is_strict_scope(None):
1917 return None
1918 if not isinstance(snapshot, dict):
1919 logger.bind(policy_audit=True).warning(
1920 "Disabling include_full_content: settings snapshot "
1921 "is missing or malformed"
1922 )
1923 self.include_full_content = False
1924 return None
1925 try:
1926 from ..security.egress.policy import (
1927 PolicyDeniedError,
1928 context_from_snapshot,
1929 resolve_run_primary_engine,
1930 )
1931 except ImportError:
1932 logger.debug(
1933 "egress_policy unavailable in search_engine_base; "
1934 "FullSearchResults fetches will be SSRF-only"
1935 )
1936 return None
1938 from ..search_system import username_from_snapshot
1940 try:
1941 # Thread username so a per-user private retriever primary
1942 # resolves PRIVATE_ONLY here too; otherwise the per-URL fetch
1943 # gate would fall back to BOTH and allow public hosts under a
1944 # run whose posture is private.
1945 primary_default = (
1946 None if _is_strict_scope(snapshot) else self._engine_name
1947 )
1948 return context_from_snapshot(
1949 snapshot,
1950 resolve_run_primary_engine(snapshot, default=primary_default),
1951 username=username_from_snapshot(snapshot),
1952 )
1953 except (PolicyDeniedError, ValueError) as exc:
1954 safe_msg = self._scrub_error(exc)
1955 logger.bind(policy_audit=True).warning(
1956 "Disabling include_full_content: egress policy could "
1957 "not be evaluated for this engine",
1958 reason=safe_msg,
1959 )
1960 self.include_full_content = False
1961 return None
1963 def set_fetch_progress_callback(self, hook) -> None:
1964 """Attach a ``(done, total)`` bulk-fetch liveness hook.
1966 Stores the hook for future ``_init_full_search`` calls and, when
1967 this engine already holds a ``FullSearchResults`` wrapper (legacy
1968 ``include_full_content`` seam), forwards it immediately. Used by
1969 the factory stamp and by ``AdvancedSearchSystem`` late-binding so
1970 pipeline-strategy full-search progress works regardless of whether
1971 the engine was created before the UI callback existed.
1972 """
1973 self._fetch_progress_callback = hook
1974 full_search = getattr(self, "full_search", None)
1975 if full_search is not None and hasattr(
1976 full_search, "progress_callback"
1977 ):
1978 full_search.progress_callback = hook
1980 def _init_full_search(
1981 self,
1982 web_search=None,
1983 language="en",
1984 max_results=10,
1985 region=None,
1986 time_period=None,
1987 safe_search=None,
1988 progress_callback=None,
1989 ):
1990 """Initialize FullSearchResults if include_full_content is True.
1992 Call this at the end of your __init__ after setting up your search wrapper.
1994 Args:
1995 web_search: The search wrapper/engine to pass to FullSearchResults
1996 language: Language for search results
1997 max_results: Maximum number of results
1998 region: Region/country code for results
1999 time_period: Time period filter
2000 safe_search: Safe search setting (string value for FullSearchResults)
2001 progress_callback: Optional ``(done, total)`` liveness hook for
2002 the bulk page fetch. Defaults to the hook stored on
2003 ``self`` (construction-time or ``set_fetch_progress_callback``).
2004 """
2005 if self.include_full_content and self.llm: 2005 ↛ exitline 2005 didn't return from function '_init_full_search' because the condition on line 2005 was always true
2006 try:
2007 from .engines.full_search import FullSearchResults
2009 # Egress context for per-URL scope gating. Built once
2010 # here (not per fetch) so the policy snapshot is the
2011 # same one used by the factory PEP that authorized this
2012 # engine.
2013 egress_context = self._build_full_search_egress_context()
2014 self.full_search = FullSearchResults(
2015 llm=self.llm,
2016 web_search=web_search,
2017 language=language,
2018 max_results=max_results,
2019 region=region,
2020 time=time_period,
2021 safesearch=safe_search,
2022 settings_snapshot=self.settings_snapshot,
2023 egress_context=egress_context,
2024 # Same hand-over as the factory's
2025 # ``_create_full_search_wrapper``: the engine's own
2026 # explicit boolean grant, nothing else, so an engine
2027 # built through this seam does not lose it silently.
2028 allow_private_ips=self.allow_private_result_fetch is True,
2029 content_cache=self.get_content_cache(),
2030 progress_callback=(
2031 progress_callback
2032 if progress_callback is not None
2033 else self._fetch_progress_callback
2034 ),
2035 )
2036 except ImportError:
2037 logger.warning(
2038 "FullSearchResults not available. "
2039 "Full content retrieval disabled."
2040 )
2041 self.include_full_content = False
2043 def _temp_attributes(self):
2044 """Return list of temporary attribute names to clean up after run().
2046 Override in subclasses that store additional temporary data.
2047 """
2048 return ["_search_results"]
2050 def _sanitize_error_message(self, message: str) -> str:
2051 """
2052 Remove/mask API keys, tokens, and secrets from error messages.
2054 Uses pattern matching for common credential formats.
2056 Args:
2057 message: The error message to sanitize
2059 Returns:
2060 Sanitized message with sensitive data redacted
2062 Example:
2063 >>> engine._sanitize_error_message(
2064 ... "Error with key sk-abc123def456ghi789jkl012"
2065 ... )
2066 "Error with key [REDACTED_KEY]"
2067 """
2068 return sanitize_error_message(message)
2070 def _scrub_error(self, error: Union[BaseException, str]) -> str:
2071 """Return a log/DB-safe rendering of *error*.
2073 Delegates the "triple-scrub" to :func:`~..security.log_sanitizer.scrub_error`,
2074 resolving this engine's known literal secret values from
2075 ``_secret_attrs``. *error* may be an exception or a pre-built
2076 message string.
2078 Use this at every catch site that logs or persists an exception so
2079 the scrub passes can never drift apart per-engine.
2080 """
2081 return scrub_error(
2082 error, *(getattr(self, name, None) for name in self._secret_attrs)
2083 )
2085 def _mask_api_key(self, api_key: str, visible_chars: int = 4) -> str:
2086 """
2087 Mask an API key for safe logging, showing only first and last characters.
2089 Args:
2090 api_key: The API key to mask
2091 visible_chars: Number of characters to show at start and end
2093 Returns:
2094 Masked API key in format "sk-1...nop" or "***" for short keys
2096 Example:
2097 >>> engine._mask_api_key("sk-abcdefghijklmnop123456")
2098 "sk-a...3456"
2099 >>> engine._mask_api_key("short")
2100 "***"
2101 """
2102 if not api_key:
2103 return "***"
2105 api_key = str(api_key).strip()
2107 if len(api_key) <= visible_chars * 2:
2108 return "***"
2110 return f"{api_key[:visible_chars]}...{api_key[-visible_chars:]}"
2112 @abstractmethod
2113 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
2114 """
2115 Get preview information (titles, summaries) for initial search results.
2117 Args:
2118 query: The search query
2120 Returns:
2121 List of preview dictionaries with at least 'id', 'title', and 'snippet' keys
2122 """
2123 pass
2125 def close(self) -> None:
2126 """
2127 Close any resources held by this search engine.
2129 Subclasses with HTTP sessions or other resources should override this.
2130 The base implementation safely closes any 'session' attribute if present
2131 and closes both preview and content filters that hold resources.
2132 """
2133 from ..utilities.resource_utils import safe_close
2135 if hasattr(self, "session") and self.session is not None:
2136 safe_close(self.session, "HTTP session")
2137 # Close preview filters as well as content filters — the journal
2138 # reputation filter is registered as a preview filter on academic
2139 # engines (arxiv, pubmed, openalex, nasa_ads, semantic_scholar)
2140 # and holds a SearXNG engine + LLM client that need releasing.
2141 if hasattr(self, "_preview_filters"):
2142 for preview_filter in self._preview_filters: 2142 ↛ 2143line 2142 didn't jump to line 2143 because the loop on line 2142 never started
2143 safe_close(preview_filter, "preview filter")
2144 if hasattr(self, "_content_filters"):
2145 for content_filter in self._content_filters: 2145 ↛ 2146line 2145 didn't jump to line 2146 because the loop on line 2145 never started
2146 safe_close(content_filter, "content filter")
2148 def __enter__(self):
2149 """Support context manager usage."""
2150 return self
2152 def __exit__(self, exc_type, exc_val, exc_tb):
2153 """Cleanup on context exit."""
2154 self.close()
2155 return False