Coverage for src/local_deep_research/advanced_search_system/tools/fetch/__init__.py: 96%
485 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""Agent-facing ``fetch_content`` tool builders.
3Public API:
4 FETCH_MODES — tuple of valid mode strings.
5 build_fetch_tool() — returns a LangChain ``@tool`` (or ``None`` when
6 mode == "disabled" so the caller can skip
7 registration).
9Modes:
10 disabled — fetch tool is not registered with the agent.
11 full — return the full extracted page text (legacy
12 behavior; can flood small-model context with
13 boilerplate / metadata enrichment).
14 summary_focus — LLM extracts only spans relevant to a focus
15 question the agent supplies per call.
16 summary_focus_query — same as above, but the prompt also includes
17 the original research query (passed in
18 programmatically by the strategy) so the
19 extractor can disambiguate vague focuses.
21Each tool registers fetched URLs in the strategy's
22``SearchResultsCollector`` for citation tracking, returning the result as
23``[N] Title: ...\nURL: ...\n\n<body>`` exactly like the original
24in-strategy implementation, so downstream prompt formatting is unchanged.
25"""
27from __future__ import annotations
29from typing import Any
31from langchain_core.language_models import BaseChatModel
32from langchain_core.tools import tool
33from loguru import logger
35from local_deep_research.utilities.js_rendering import (
36 read_js_rendering_setting as _read_js_rendering_setting,
37)
38from local_deep_research.security import (
39 redact_url_for_log,
40 sanitize_error_for_agent,
41)
42from local_deep_research.settings.exceptions import SettingsValueError
43from local_deep_research.utilities.type_utils import (
44 to_positive_int,
45 unwrap_setting,
46)
47from local_deep_research.utilities.llm_utils import invoke_llm_sync
49from .library_resolver import (
50 is_citation_reference,
51 make_library_resolver,
52 resolve_citation_reference as _resolve_citation_reference,
53)
54from .prompts import SUMMARY_FOCUS_PROMPT, SUMMARY_FOCUS_QUERY_PROMPT
57# Per-call timeouts and caps. Kept here rather than in the strategy file
58# because they are properties of the fetch tool, not of agent
59# orchestration.
60CONTENT_FETCH_TIMEOUT = 30
61CONTENT_MAX_LENGTH = 10_000
62SUMMARY_INPUT_MAX_LENGTH = 500_000
64# Bounds for the user-configurable ``search.fetch.content_max_length``
65# setting in full mode (see default_settings.json). Summary modes have a
66# separate input budget with no fixed upper setting limit.
67CONTENT_MAX_LENGTH_MIN = 1_000
68CONTENT_MAX_LENGTH_MAX = 50_000
70# The summary prompt limit is counted in characters, estimated at four per
71# token. That rate fits English-like text. Text that needs more tokens per
72# character, such as Chinese or Japanese or long runs of digits, can still
73# exceed the window.
74SUMMARY_INPUT_CHARS_PER_TOKEN = 4
76# Fixed text: it never echoes the configured value. It names the
77# environment variable because an override there takes precedence and is
78# not editable in the web UI.
79SUMMARY_INPUT_MAX_LENGTH_ERROR = (
80 "search.fetch.summary_input_max_length must be a positive integer. "
81 "Check the setting and any LDR_SEARCH_FETCH_SUMMARY_INPUT_MAX_LENGTH "
82 "environment override."
83)
86FETCH_MODES = (
87 "disabled",
88 "full",
89 "summary_focus",
90 "summary_focus_query",
91)
93# Default call budget for fetch_content per subagent (and lead agent) per topic.
94# Prevents autonomous agents from exhausting recursion limits on fetches.
95DEFAULT_MAX_FETCHES_PER_TOPIC = 2
98def _read_max_fetches_setting(
99 settings_snapshot: dict | None,
100 default: int = DEFAULT_MAX_FETCHES_PER_TOPIC,
101) -> int:
102 """Read the ``search.fetch.max_per_topic`` setting from a settings snapshot.
104 Retained as a fallback for direct callers of ``build_fetch_tool`` outside
105 ``LangGraphAgentStrategy`` (which passes ``max_fetches`` explicitly).
107 Returns the configured non-negative integer limit, or default if it is
108 missing or an untyped value is not a usable limit. Negative values fall
109 back to default, and values above 50 are clamped to 50. 0 means unlimited
110 fetches.
112 Typed number rows use the canonical accessor's float conversion (fractional
113 strings truncate) and environment priority. A non-empty
114 ``LDR_SEARCH_FETCH_MAX_PER_TOPIC`` overrides such a row. A typed row or an
115 override that is not a finite number (a boolean included), or a row whose
116 type is unsupported, propagates SettingsValueError instead of selecting
117 another limit. A negative resulting limit falls back to the caller's
118 default.
120 Raw values and sparse legacy rows retain snapshot-only int() coercion.
121 An absent exact key keeps the caller's default even with an environment or
122 thread-local value. Explicit ``max_fetches`` bypasses this fallback reader.
123 """
124 key = "search.fetch.max_per_topic"
125 # Missing snapshots keep the caller's default, not a thread-local value.
126 if not settings_snapshot or key not in settings_snapshot:
127 return default
128 from local_deep_research.config.thread_settings import (
129 get_setting_from_snapshot,
130 )
132 lookup_snapshot = settings_snapshot
133 entry = settings_snapshot[key]
134 if (
135 isinstance(entry, dict)
136 and "value" in entry
137 and "ui_element" not in entry
138 ):
139 # Sparse legacy rows have no UI type. Preserve their int() coercion
140 # instead of letting the accessor infer text for floats/booleans.
141 raw = unwrap_setting(entry)
142 if isinstance(raw, dict):
143 return default
144 lookup_snapshot = {key: raw}
146 try:
147 val = get_setting_from_snapshot(
148 key, default=None, settings_snapshot=lookup_snapshot
149 )
150 except SettingsValueError:
151 raise
152 except Exception:
153 return default
154 if val is None:
155 return default
156 unwrapped = unwrap_setting(val)
157 try:
158 limit = int(unwrapped)
159 if limit < 0:
160 return default
161 return min(50, limit)
162 except (TypeError, ValueError, OverflowError):
163 return default
166def _read_content_max_length_setting(
167 settings_snapshot: dict | None,
168 default: int = CONTENT_MAX_LENGTH,
169) -> int:
170 """Read the ``search.fetch.content_max_length`` setting from a settings snapshot.
172 The value caps page text (HTTP or local library document) before it is
173 returned directly in ``full`` mode. Summary modes use their own input
174 budget instead.
176 Returns the configured length clamped to
177 [CONTENT_MAX_LENGTH_MIN, CONTENT_MAX_LENGTH_MAX]. Missing, non-positive,
178 or unconvertible untyped legacy values retain the default. Invalid managed
179 settings propagate SettingsValueError instead of selecting another limit.
180 """
181 if not settings_snapshot:
182 return default
183 from local_deep_research.config.thread_settings import (
184 get_setting_from_snapshot,
185 )
187 try:
188 val = get_setting_from_snapshot(
189 "search.fetch.content_max_length",
190 default=None,
191 settings_snapshot=settings_snapshot,
192 )
193 except SettingsValueError:
194 raise
195 except Exception:
196 return default
197 if val is None:
198 return default
199 unwrapped = unwrap_setting(val)
200 try:
201 if isinstance(unwrapped, str):
202 stripped = unwrapped.strip()
203 try:
204 limit = int(stripped)
205 except ValueError:
206 # Allow float strings like "10000.7" (truncate toward zero,
207 # matching int(float) semantics for numeric inputs).
208 limit = int(float(stripped))
209 else:
210 limit = int(unwrapped)
211 if limit <= 0:
212 return default
213 return min(CONTENT_MAX_LENGTH_MAX, max(CONTENT_MAX_LENGTH_MIN, limit))
214 except (TypeError, ValueError, OverflowError):
215 return default
218def _read_summary_input_max_length_setting(
219 settings_snapshot: dict | None,
220) -> int:
221 """Read the positive summary-input character budget without an upper clamp.
223 Missing settings use the generous default. Invalid configured values
224 raise rather than silently increasing a user's intended budget. Settings
225 loads keep an invalid value, including an invalid environment override,
226 unchanged, so this per-fetch read is where it fails. Zero is not an
227 unlimited sentinel: HTTP and library slicing treat it differently.
228 """
229 from local_deep_research.config.thread_settings import (
230 get_setting_from_snapshot,
231 )
233 value = unwrap_setting(
234 get_setting_from_snapshot(
235 "search.fetch.summary_input_max_length",
236 default=SUMMARY_INPUT_MAX_LENGTH,
237 settings_snapshot=settings_snapshot,
238 )
239 )
240 try:
241 return to_positive_int(value)
242 except ValueError:
243 pass
244 raise ValueError(SUMMARY_INPUT_MAX_LENGTH_ERROR)
247def _built_model_provider(
248 model: Any, settings_snapshot: dict | None
249) -> tuple[str | None, dict | None]:
250 """Return the provider and settings ``get_llm`` built *model* with.
252 ``get_llm`` records both on the ``ProcessingLLMWrapper`` it returns,
253 after any per-run provider override, so they describe the model that
254 runs. Any other model has no known provider (``None``), and the
255 tool's own snapshot stands in for its settings.
256 """
257 from local_deep_research.config.llm_config import ProcessingLLMWrapper
258 from local_deep_research.llm.providers.base import normalize_provider
260 if not isinstance(model, ProcessingLLMWrapper):
261 return None, settings_snapshot
262 provider = getattr(model, "_provider", None)
263 snapshot = getattr(model, "_settings_snapshot", None)
264 if not isinstance(snapshot, dict): 264 ↛ 265line 264 didn't jump to line 265 because the condition on line 264 was never true
265 snapshot = settings_snapshot
266 # Registry overrides can use a built-in provider's name for an arbitrary
267 # shared instance or factory. Its invocation contract and context are not
268 # those of the provider LDR builds, even though get_llm wraps it.
269 if model._owns_llm is False:
270 return None, snapshot
271 if not isinstance(provider, str) or not provider.strip(): 271 ↛ 272line 271 didn't jump to line 272 because the condition on line 271 was never true
272 return None, snapshot
273 return normalize_provider(provider.strip()), snapshot
276def _summary_invocation_budget(
277 model: Any, settings_snapshot: dict | None
278) -> tuple[int | None, dict[str, int]]:
279 """Return the summary prompt character limit and per-call model options.
281 ``None`` means no limit: the model was built for a cloud provider with
282 unrestricted context, which enforces its own limit.
284 Otherwise the limit is the context window minus the response
285 allowance, in tokens, at ``SUMMARY_INPUT_CHARS_PER_TOKEN`` characters
286 per token. The whole prompt counts, so the page gets what its
287 instructions, focus, query, title and URL leave.
289 The window is resolved with ``get_context_window_for_provider``, as
290 the built-in providers' builds resolve it, from the provider and
291 settings ``get_llm`` used (see ``_built_model_provider``):
293 - Ollama: ``llm.local_context_window_size``, sent as ``num_ctx``.
294 - LM Studio and llama.cpp: ``llm.local_context_window_size``. It caps
295 their ``max_tokens`` but is not sent; each server keeps its own
296 context length.
297 - Any other provider: ``llm.context_window_size`` when
298 ``llm.context_window_unrestricted`` is off.
300 When that resolves no window and the provider is not a known cloud
301 provider (the egress policy's ``_CLOUD_LLM_PROVIDERS``),
302 ``llm.local_context_window_size`` is assumed. That covers an
303 OpenAI- or Anthropic-compatible endpoint, which may be local or
304 private, a provider LDR does not ship, and a model ``get_llm`` did not
305 build. A window that is unusable or below 1 falls back to that local
306 window, then to ``DEFAULT_LOCAL_CONTEXT_WINDOW_SIZE``.
308 The response allowance is ``compute_max_tokens`` for that window:
309 ``llm.max_tokens``, capped at ``MAX_TOKENS_CONTEXT_SHARE`` (80%) of
310 it, which is the ``max_tokens`` OpenAI-compatible and Anthropic builds
311 request. When that yields no positive value, for example with
312 ``llm.supports_max_tokens`` off, the full 80% share is kept. Ollama is
313 sent no ``max_tokens``, but its reply must fit in ``num_ctx`` too, so
314 it keeps the same allowance.
316 An unrestricted compatible endpoint's existing model may request more
317 output than the assumed local window permits. Pass the capped allowance
318 through the existing model wrappers for this summary call only; do not
319 mutate the shared model or bind around its rate limiter.
320 """
321 from local_deep_research.config.thread_settings import (
322 NoSettingsContextError,
323 )
324 from local_deep_research.llm.providers._helpers import (
325 DEFAULT_LOCAL_CONTEXT_WINDOW_SIZE,
326 MAX_TOKENS_CONTEXT_SHARE,
327 compute_max_tokens,
328 get_context_window_for_provider,
329 get_local_context_window,
330 )
331 from local_deep_research.security.egress.policy import (
332 _CLOUD_LLM_PROVIDERS,
333 )
335 unusable = (TypeError, ValueError, OverflowError, NoSettingsContextError)
336 provider, snapshot = _built_model_provider(model, settings_snapshot)
337 try:
338 window = get_context_window_for_provider(
339 provider, settings_snapshot=snapshot
340 )
341 except SettingsValueError:
342 raise
343 except unusable:
344 window = None
345 else:
346 if window is None and provider in _CLOUD_LLM_PROVIDERS:
347 return None, {}
348 local_window_assumed = window is None or window < 1
349 if local_window_assumed:
350 try:
351 window = get_local_context_window(snapshot)
352 except SettingsValueError:
353 raise
354 except unusable:
355 window = None
356 if window is None or window < 1:
357 window = DEFAULT_LOCAL_CONTEXT_WINDOW_SIZE
359 try:
360 allowance = compute_max_tokens(
361 settings_snapshot=snapshot, context_window_size=window
362 )
363 except SettingsValueError:
364 raise
365 except unusable:
366 allowance = None
367 invocation_kwargs = {}
368 if not isinstance(allowance, int) or allowance < 1:
369 allowance = int(window * MAX_TOKENS_CONTEXT_SHARE)
370 elif local_window_assumed and provider in (
371 "openai_endpoint",
372 "anthropic_endpoint",
373 ):
374 invocation_kwargs["max_tokens"] = allowance
375 return (
376 max(1, (window - allowance) * SUMMARY_INPUT_CHARS_PER_TOKEN),
377 invocation_kwargs,
378 )
381_warned_collector_missing_fetch_cap = False
384def _warn_collector_lacks_fetch_recording() -> None:
385 global _warned_collector_missing_fetch_cap
386 if not _warned_collector_missing_fetch_cap:
387 _warned_collector_missing_fetch_cap = True
388 logger.warning(
389 "Collector does not implement try_record_fetch; fetch content budget cap is disabled."
390 )
393def _check_and_record_fetch(
394 collector: Any,
395 max_fetches: int,
396 subagent_id: str | None = None,
397) -> tuple[str | None, int | None]:
398 """Check if the fetch budget has been exhausted. If not, record the fetch attempt.
400 Requires ``try_record_fetch`` on the collector (provided by
401 ``SearchResultsCollector``). If collector is None, budget enforcement is skipped.
402 If collector does not implement ``try_record_fetch``, a one-time warning is logged.
404 Accounting: Only invocations that reach the network consume a slot. Pre-resolved
405 local library reads and egress policy denials do not consume budget slots.
407 Returns:
408 (budget_err, current_count)
409 budget_err is an informative error string if budget is exhausted, or None if the fetch can proceed.
410 current_count is the current fetch count for this subagent/topic, or None.
411 """
412 if collector is not None and hasattr(collector, "try_record_fetch"):
413 try:
414 allowed, count = collector.try_record_fetch(
415 max_fetches, subagent_id=subagent_id
416 )
417 except TypeError:
418 allowed, count = collector.try_record_fetch(max_fetches)
419 if not allowed:
420 logger.warning(
421 f"fetch_content call budget exhausted ({count}/{max_fetches}) for this topic"
422 )
423 return (
424 f"Fetch content budget exhausted ({count}/{max_fetches}) for this topic. "
425 "Rely on search snippets or synthesize your findings with the information already gathered.",
426 count,
427 )
428 return None, count
429 if collector is not None: 429 ↛ 431line 429 didn't jump to line 431 because the condition on line 429 was always true
430 _warn_collector_lacks_fetch_recording()
431 return None, None
434def _register_in_collector(
435 collector: Any,
436 url: str,
437 title: str,
438 snippet_source: str,
439 source_metadata: dict | None = None,
440) -> int:
441 """Register a fetched URL in the collector and return its 1-based citation index.
443 If the URL was already tracked (via a prior search hit) the existing
444 index is reused so the agent sees a stable citation per URL.
446 .. note::
447 The ``find_or_add_result``-less branch below is **speculative
448 generality**: ``SearchResultsCollector`` is the only collector in
449 this repo and it has that method, so nothing in-tree reaches the
450 fallback and no test exercises it against a real collector. It
451 exists for third-party or pre-#5363 collectors. Treat it as
452 unverified against any live caller — do not assume a bug there is
453 affecting production, and do not grow it further without a real
454 consumer to justify it.
456 This branch was left at its pre-existing shape deliberately. An
457 earlier revision of this follow-up grew it — duck-typed tuple
458 unpacking for a bare-``int`` ``add_results`` contract, a ``getattr``
459 guard on ``find_by_url`` — for a collector that does not exist in
460 any repo we can see, which is precisely what the paragraph above
461 forbids. That growth was reverted rather than shipped.
462 """
463 metadata = _complete_archive_metadata(url, source_metadata) or {}
464 source_engine = (
465 "wayback" if metadata.get("source_type") == "archive" else "fetch"
466 )
467 if metadata.get("source_type") == "archive":
468 title = f"{title or metadata['original_url']} (archived {metadata['formatted_date']})"
469 snippet = snippet_source[:200].strip()
470 if len(snippet_source) > 200:
471 snippet += "..."
472 if not hasattr(collector, "find_or_add_result"):
473 # Legacy/custom collectors without find_or_add_result fallback
474 _, indexed = collector.add_results(
475 [{**metadata, "title": title, "link": url, "snippet": snippet}],
476 engine_name=source_engine,
477 )
478 if indexed: 478 ↛ 488line 478 didn't jump to line 488 because the condition on line 478 was always true
479 raw_idx = indexed[0].get("index")
480 try:
481 return int(raw_idx)
482 except (ValueError, TypeError):
483 logger.warning(
484 "Failed to parse citation index for url={}: {!r}",
485 redact_url_for_log(url),
486 raw_idx,
487 )
488 existing_idx = collector.find_by_url(url)
489 if existing_idx is not None:
490 return existing_idx
491 raise RuntimeError(
492 f"Failed to register fetched URL in collector: {redact_url_for_log(url)}"
493 )
495 return collector.find_or_add_result(
496 {**metadata, "title": title, "link": url, "snippet": snippet},
497 engine_name=source_engine,
498 )
501def _format_fetch_message(cite_idx, title, url, content, source_metadata):
502 message = f"[{cite_idx}] Title: {title}\nURL: {url}\n\n{content}"
503 if source_metadata and source_metadata.get("source_type") == "archive":
504 message += (
505 f"\n\nOriginal URL: {source_metadata['original_url']}"
506 f"\nCaptured: {source_metadata['formatted_date']}"
507 )
508 requested = source_metadata.get("requested_url")
509 if requested and requested != source_metadata["original_url"]:
510 message += f"\nRequested URL: {requested}"
511 return message
514def _enforce_url_policy(url: str, egress_context: Any) -> None:
515 """Run ``evaluate_url`` against ``egress_context`` and raise
516 ``PolicyDeniedError`` on denial.
518 No-op when no context is configured (callers without policy enforcement,
519 e.g. legacy non-LangGraph strategies, see the legacy behavior).
520 """
521 if egress_context is None:
522 return
523 from local_deep_research.security.egress.policy import (
524 PolicyDeniedError,
525 evaluate_url,
526 )
528 decision = evaluate_url(url, egress_context)
529 if not decision.allowed:
530 raise PolicyDeniedError(decision, target=url)
533def _denial_reason(exc: Any) -> str:
534 """Best-effort egress-denial reason code for an agent-facing message."""
535 return getattr(getattr(exc, "decision", None), "reason", "policy_denied")
538def _is_fetch_cache_hit_allowed(
539 url: str, egress_context: Any, _cache: Any | None = None
540) -> bool:
541 """Whether a shared-cache hit for *url* may be served under *egress_context*.
543 Defense-in-depth for the shared research cache: ``_enforce_url_policy``
544 (``evaluate_url``, scope-only) already passed at this point, but the
545 uncached path still runs ``ContentFetcher``'s SSRF gate
546 (``policy_aware_validate_url``) which refuses private/loopback hosts
547 unless the run holds a private grant (``PRIVATE_ONLY``). The cache fast
548 path must apply the same SSRF gate before serving — otherwise an
549 ``UNPROTECTED``-scope run (where ``evaluate_url`` allows any host)
550 would serve a cached private entry that the fetch path refuses.
552 Library routes never reach here as cache hits: valid references return
553 early via ``_try_resolve_url`` (local DB read, no HTTP), and anything
554 else fails this check closed (``validate_url`` rejects non-http schemes
555 and unresolvable ``library.document`` aliases).
557 Fail-closed: when the validator cannot be imported the hit is refused
558 (callers fall through to ``ContentFetcher``, which enforces its own
559 gate). When *_cache* is a ``ResearchContentCache`` successful verdicts
560 are memoized per-run so repeat hits skip DNS and stay servable when
561 DNS drops mid-run after a success; refusals (including import-failure
562 and transient DNS failures) are never pinned, so recovery stays open.
563 """
565 def _compute() -> bool:
566 try:
567 from local_deep_research.security.egress.fetch import (
568 policy_aware_validate_url,
569 )
570 except Exception: # pragma: no cover - defensive
571 logger.debug(
572 "fetch cache grant check: validator unavailable, refusing hit"
573 )
574 return False
575 try:
576 return bool(policy_aware_validate_url(url, egress_context))
577 except Exception: # pragma: no cover - defensive
578 return False
580 # Per-run memo keyed by the exact SSRF booleans the compute uses:
581 # PRIVATE_ONLY maps to (allow_private=True, block_link_local=True),
582 # mirroring the fetcher wrapper; every other scope/None maps to the
583 # strict defaults (False, False). Only True is pinned — a False only
584 # skips the cache fast path while the fetcher re-validates, so a
585 # transient DNS blip or import failure never pins the URL.
586 try:
587 _memo = getattr(_cache, "validated", None)
588 if callable(_memo): 588 ↛ 610line 588 didn't jump to line 610 because the condition on line 588 was always true
589 try:
590 from local_deep_research.security.egress.policy import (
591 EgressScope,
592 )
593 except Exception: # pragma: no cover - defensive
594 EgressScope = None # type: ignore[assignment]
595 allow_private = bool(
596 EgressScope is not None
597 and egress_context is not None
598 and getattr(egress_context, "scope", None)
599 == EgressScope.PRIVATE_ONLY
600 )
601 block_link_local = bool(allow_private)
602 try:
603 return bool(
604 _memo(url, allow_private, block_link_local, _compute)
605 )
606 except Exception: # pragma: no cover - defensive
607 return False
608 except Exception: # pragma: no cover - defensive
609 logger.debug("fetch cache grant check: memo lookup failed")
610 return _compute()
613# Shared instruction appended to every per-URL egress denial returned to the
614# agent. Tells it WHY the fetch was refused and what to do instead, so it
615# adapts (stays in-scope) rather than retrying the same out-of-scope URL.
616_EGRESS_DENIAL_HINT = (
617 "In this run only local collection/library documents can be fetched; "
618 "skip external URLs."
619)
621# ---------------------------------------------------------------------------
622# Pre-resolution: rewrite non-network URLs BEFORE the egress gate.
623#
624# Two URL-shaped strings reach the fetch tool that aren't actually network
625# fetches and that the egress policy correctly rejects as
626# ``unsupported_scheme``. Resolving them here lets the agent get the page
627# content (or a helpful error) instead of a generic denial — A3 from
628# research_f3045c5b_issue_analysis.md.
629# ---------------------------------------------------------------------------
631# ``_KIND_RESULT`` marks a pre-resolved local document; the fetch tool
632# consumes the dict directly. ``_KIND_ERROR`` short-circuits the fetch
633# tool with the message in ``payload`` (already routed through the
634# credential scrubber by the caller). ``_KIND_REWRITTEN`` means "this
635# URL was rewritten to a new one — use the new URL going forward but
636# keep running the normal HTTP flow" (e.g. a ``[N]`` citation marker
637# that resolves to an external URL). ``None`` means "no pre-resolution;
638# run the normal HTTP path with the original URL."
639_KIND_RESULT = "result"
640_KIND_ERROR = "error"
641_KIND_REWRITTEN = "rewritten"
644def _try_resolve_url(
645 url: str,
646 library_resolver: Any,
647 collector: Any,
648 _visited: set[str] | None = None,
649 content_max_length: int | None = None,
650) -> tuple[str, Any] | None:
651 """Pre-resolve a fetch URL that isn't a network URL.
653 Returns one of:
655 - ``(``_KIND_RESULT``, dict)`` — local-content payload shaped like a
656 ``ContentFetcher.fetch()`` success result, ready for the post-fetch
657 pipeline. The caller MUST skip the egress gate (it's a local read).
658 - ``(``_KIND_ERROR``, str)`` — short-circuit with this error message.
659 - ``(``_KIND_REWRITTEN``, str)`` — *url* was rewritten (e.g. a
660 citation marker ``[N]`` that resolved to an external URL); use
661 the new URL but otherwise run the normal HTTP flow (the egress
662 gate still applies).
663 - ``None`` — no pre-resolution; run the normal HTTP path with *url*.
665 Handles:
667 - ``[N]`` citation markers → ``SearchResultsCollector.find_by_index``,
668 then recurse with the resolved URL (so a citation whose source is a
669 library doc still hits the local fast path, while a citation whose
670 source is external surfaces a ``_KIND_REWRITTEN`` URL for the gate).
671 - ``/library/document/<uuid>[/pdf]`` → ``Document.text_content`` read
672 from the user DB via the library resolver.
673 """
674 if not isinstance(url, str):
675 return None
677 if _visited is None:
678 _visited = set()
679 if url in _visited:
680 return (
681 _KIND_ERROR,
682 f"Circular citation reference detected for '{url}'.",
683 )
684 if len(_visited) >= 5:
685 return (
686 _KIND_ERROR,
687 f"Citation resolution depth limit exceeded for '{url}'.",
688 )
689 _visited.add(url)
691 # 1. Citation marker [N] → resolve to citation's URL.
692 citation = _resolve_citation_reference(url, collector)
693 if citation is not None:
694 resolved_url = citation.get("link") or citation.get("url") or ""
695 if not resolved_url:
696 return (
697 _KIND_ERROR,
698 (
699 f"Citation {url} has no URL field. "
700 "Use the citation's URL or a different source instead."
701 ),
702 )
703 # First, check if the citation's URL is itself a library doc
704 # (the recursive call sees the library path and returns _KIND_RESULT
705 # directly). Otherwise surface the citation's URL as a rewrite so
706 # the fetch tool updates its ``url`` variable before the egress
707 # gate / HTTP fetch runs against the resolved URL — not the marker.
708 recursive = _try_resolve_url(
709 resolved_url,
710 library_resolver,
711 collector,
712 _visited=_visited,
713 content_max_length=content_max_length,
714 )
715 if recursive is not None:
716 return recursive
717 return (_KIND_REWRITTEN, resolved_url)
719 # 2. Citation marker that doesn't match a tracked citation.
720 # ``_resolve_citation_reference`` only returns the citation for a
721 # well-formed ``[N]`` marker; anything else falls through to the
722 # library-resolver check below.
723 if is_citation_reference(url) is not None:
724 return (
725 _KIND_ERROR,
726 (
727 f"No registered citation matches {url}. The agent's "
728 "search results use citation markers; use the source URL "
729 "(the link next to the marker) or a tracked citation "
730 "instead of a raw marker."
731 ),
732 )
734 # 3. Library document URL.
735 if library_resolver is not None:
736 content = library_resolver(url)
737 if content is not None:
738 text = content.get("content") or ""
739 cap = (
740 content_max_length
741 if content_max_length is not None
742 else CONTENT_MAX_LENGTH
743 )
744 if len(text) > cap:
745 text = text[:cap]
746 return (
747 _KIND_RESULT,
748 {
749 "status": "success",
750 "title": content.get("title") or "",
751 "content": text,
752 "url": content.get("url") or url,
753 },
754 )
756 return None
759def _complete_archive_metadata(url: str, metadata: dict | None) -> dict | None:
760 """Recover missing provenance from a verified capture before using it."""
761 if metadata is None or metadata.get("source_type") != "archive":
762 return metadata
763 from local_deep_research.web_search_engines.wayback_client import (
764 capture_metadata_from_replay_url,
765 validate_capture_fetch_url,
766 )
768 captured = capture_metadata_from_replay_url(url)
769 if captured is None:
770 raise ValueError("Archive provenance requires an exact replay URL")
771 # Missing metadata is recoverable, but supplied constraints must not
772 # be replaced silently when they disagree with the actual capture.
773 timestamp = metadata.get("timestamp")
774 original_url = metadata.get("original_url")
775 if timestamp is None or timestamp == "":
776 timestamp = captured["timestamp"]
777 if original_url is None or original_url == "":
778 original_url = captured["original_url"]
779 if not isinstance(timestamp, str) or not isinstance(original_url, str): 779 ↛ 780line 779 didn't jump to line 780 because the condition on line 779 was never true
780 raise ValueError("Invalid archive provenance")
781 validate_capture_fetch_url(url, timestamp, original_url)
782 return {
783 **metadata,
784 **captured,
785 "requested_url": metadata.get("requested_url")
786 or captured["requested_url"],
787 }
790def _archive_fetch_context(
791 url: str,
792 collector: Any,
793 source_metadata: dict | None,
794 url_guard: Any,
795) -> tuple[dict | None, Any]:
796 """Retain a tracked capture's provenance when fetching it again.
798 Resolve both citation chains and literal replay links using the
799 collector's persistent index, including after a subsection reset.
800 The resulting guard stays local to this call.
801 """
802 metadata = source_metadata
803 if is_citation_reference(url) is not None:
804 # Use the HTTP path's cycle/depth checks to reach the final target.
805 # Leave local document reads to that path's actual library resolver.
806 resolved = _try_resolve_url(url, None, collector)
807 if resolved is None or resolved[0] != _KIND_REWRITTEN:
808 return metadata, url_guard
809 url = resolved[1]
810 tracked = None
811 if metadata is None and collector is not None:
812 find_url = getattr(collector, "find_by_url", None)
813 find_index = getattr(collector, "find_by_index", None)
814 if callable(find_url) and callable(find_index):
815 index = find_url(url)
816 if isinstance(index, int):
817 tracked = find_index(index)
818 if (
819 not isinstance(tracked, dict)
820 or tracked.get("source_type") != "archive"
821 ):
822 replay = (
823 tracked.get("link") or tracked.get("url")
824 if isinstance(tracked, dict)
825 else url
826 )
827 find_archive = getattr(collector, "find_archive_by_url", None)
828 if isinstance(replay, str) and callable(find_archive):
829 archived = find_archive(replay)
830 if isinstance(archived, dict): 830 ↛ 831line 830 didn't jump to line 831 because the condition on line 830 was never true
831 tracked = archived
832 if (
833 isinstance(tracked, dict)
834 and tracked.get("source_type") == "archive"
835 ):
836 metadata = {
837 field: tracked[field]
838 for field in (
839 "source_type",
840 "original_url",
841 "requested_url",
842 "timestamp",
843 "formatted_date",
844 )
845 if field in tracked
846 }
847 from local_deep_research.web_search_engines.wayback_client import (
848 capture_metadata_from_replay_url,
849 validate_capture_fetch_url,
850 )
852 candidate = (
853 tracked.get("link") or tracked.get("url")
854 if collector is not None and isinstance(tracked, dict)
855 else url
856 )
857 if metadata is None:
858 if isinstance(candidate, str): 858 ↛ 860line 858 didn't jump to line 860 because the condition on line 858 was always true
859 metadata = capture_metadata_from_replay_url(candidate)
860 if metadata is None or metadata.get("source_type") != "archive":
861 return metadata, url_guard
862 metadata = _complete_archive_metadata(candidate, metadata)
864 def capture_guard(candidate):
865 validate_capture_fetch_url(
866 candidate, metadata["timestamp"], metadata["original_url"]
867 )
868 if url_guard is not None:
869 url_guard(candidate)
871 return metadata, capture_guard
874def _archive_failure_message(exc, availability, source_metadata):
875 """Classify archive outages without disabling a live fetch provider."""
876 if availability is None or not source_metadata:
877 return None
878 if source_metadata.get("source_type") != "archive": 878 ↛ 879line 878 didn't jump to line 879 because the condition on line 878 was never true
879 return None
880 from local_deep_research.web_search_engines.engine_availability import (
881 failure_from_exception,
882 )
884 failure = failure_from_exception(exc)
885 if failure is not None:
886 return availability.disable("wayback", failure).describe()
887 return None
890def _fetch_raw_content(
891 url: str,
892 library_resolver: Any,
893 collector: Any,
894 egress_context: Any,
895 settings_snapshot: dict | None,
896 mode_label: str,
897 max_fetches: int | None = None,
898 subagent_id: str | None = None,
899 content_cache: Any = None,
900 content_max_length: int | None = None,
901 url_guard: Any = None,
902 availability: Any = None,
903 source_metadata: dict | None = None,
904) -> tuple[dict | None, str, str | None, int | None]:
905 """Pre-resolve non-network URL shapes or fetch via HTTP.
907 ``content_cache`` is the single shared cache for the whole research
908 run (quick phase, subsections, lead + subagents). When None, the
909 cache bound to the current research (if any) is used via
910 ``ContentFetcher``'s own registry fallback. Repeat URLs are therefore
911 served from previously fetched content instead of hitting the network
912 on the spot. A guarded archive read (``url_guard`` set) never reads or
913 stores cached page content.
915 Returns:
916 (result, final_url, error_string, current_count)
917 If error_string is not None, a pre-resolution error or budget exhaustion
918 occurred (already scrubbed) and should be returned immediately by the fetch tool.
919 current_count is the current fetch count if a network fetch was recorded, or None.
920 """
921 from local_deep_research.content_fetcher import ContentFetcher
922 from local_deep_research.utilities.research_content_cache import (
923 resolve_cache,
924 )
926 try:
927 _cache = resolve_cache(content_cache)
928 except Exception: # pragma: no cover - defensive
929 _cache = None
930 if url_guard is not None:
931 # A guarded (archive) read neither reads nor stores cached page
932 # content, here or in ContentFetcher: an entry under the same replay
933 # URL can come from an unguarded fetch whose redirects left the
934 # capture.
935 _cache = None
937 effective_max_length = (
938 content_max_length
939 if content_max_length is not None
940 else _read_content_max_length_setting(settings_snapshot)
941 )
942 pre = _try_resolve_url(
943 url,
944 library_resolver,
945 collector,
946 content_max_length=effective_max_length,
947 )
948 result: dict | None = None
949 if pre is not None:
950 kind, payload = pre
951 if kind == _KIND_ERROR:
952 return None, url, sanitize_error_for_agent(payload), None
953 if kind == _KIND_RESULT:
954 # A local library document was resolved directly from the user DB.
955 # Skip both the egress gate (local read) and ContentFetcher (no HTTP).
956 # It does not consume a network fetch budget slot.
957 result = payload
958 url = result.get("url") or url
959 logger.info(
960 f"[FETCH] mode={mode_label} source=library "
961 f"url={redact_url_for_log(url)} — "
962 "resolved local library document directly"
963 )
964 else: # _KIND_REWRITTEN — citation marker resolved to a URL
965 url = payload
967 current_count: int | None = None
968 if result is None:
969 # Either no pre-resolution or a citation-marker rewrite —
970 # run the normal HTTP path. Per-URL egress gate (pre-fetch)
971 # + ContentFetcher's own per-redirect gate both raise
972 # PolicyDeniedError on an out-of-scope URL.
973 # Check the egress gate BEFORE recording the fetch so that
974 # egress denials do not consume budget slots.
975 if source_metadata and source_metadata.get("source_type") == "archive":
976 if availability is not None:
977 state = availability.state("wayback")
978 if state is not None:
979 return None, url, state.describe(), None
980 if max_fetches is not None:
981 # An ordinary reread reserves its slot here. The archive
982 # wrapper has already checked these gates before lookup.
983 from local_deep_research.security.egress.policy import (
984 PolicyDeniedError,
985 evaluate_engine,
986 )
987 from local_deep_research.web_search_engines.wayback_client import (
988 AVAILABLE_API,
989 )
991 if egress_context is not None:
992 decision = evaluate_engine(
993 "wayback",
994 egress_context,
995 settings_snapshot=settings_snapshot or {},
996 )
997 if not decision.allowed:
998 raise PolicyDeniedError(decision, target=AVAILABLE_API)
999 _enforce_url_policy(
1000 source_metadata["original_url"], egress_context
1001 )
1002 url_guard(url)
1003 _enforce_url_policy(url, egress_context)
1005 # Egress check passed! Only count fetches that reach the network.
1006 if max_fetches is not None:
1007 budget_err, current_count = _check_and_record_fetch(
1008 collector, max_fetches, subagent_id=subagent_id
1009 )
1010 if budget_err is not None:
1011 return None, url, budget_err, current_count
1013 # Inspect the cache only for logging. ContentFetcher serves both
1014 # hits and misses below, applying the same policy and max_length
1015 # to both. A separate early return here would bypass changes to
1016 # the caller's content limit (including summary prompt limits).
1017 _hit = None
1018 if _cache is not None and _is_fetch_cache_hit_allowed(
1019 url, egress_context, _cache
1020 ):
1021 try:
1022 _hit = _cache.get(url)
1023 except Exception: # pragma: no cover - defensive
1024 _hit = None
1026 enable_js = _read_js_rendering_setting(settings_snapshot)
1027 fetcher_options = {
1028 "timeout": CONTENT_FETCH_TIMEOUT,
1029 "enable_js_rendering": enable_js,
1030 "egress_context": egress_context,
1031 "content_cache": _cache,
1032 }
1033 if url_guard is not None:
1034 fetcher_options["url_guard"] = url_guard
1035 with ContentFetcher(**fetcher_options) as fetcher:
1036 result = fetcher.fetch(url, max_length=effective_max_length)
1037 if (
1038 source_metadata
1039 and source_metadata.get("source_type") == "archive"
1040 and availability is not None
1041 and result.get("status") == "success"
1042 and (result.get("content") or "").strip()
1043 ):
1044 availability.record_success("wayback")
1046 if (
1047 _hit is not None
1048 and isinstance(_hit.get("content"), str)
1049 and result.get("status") == "success"
1050 ):
1051 logger.info(
1052 f"[FETCH] mode={mode_label} source=cache "
1053 f"url={redact_url_for_log(url)} — "
1054 "served from shared research cache"
1055 )
1057 # Store successes at this layer only when ContentFetcher did not
1058 # already cache the full text (e.g. mocked fetchers in tests).
1059 # Never write truncated display output back: ContentFetcher caches
1060 # the full pre-truncation text, and a 10_001-char page sliced to
1061 # 10_000 + marker is LONGER, so it would win under longest-wins
1062 # and lose the tail for later callers with larger limits.
1063 # Only successes: errors must retry. Grant-consistent: entries
1064 # this caller's grant would not allow to read are not written
1065 # either (mirrors the engine store gate).
1066 if (
1067 _cache is not None
1068 and isinstance(result, dict)
1069 and result.get("status") == "success"
1070 and isinstance(result.get("content"), str)
1071 and result.get("content_is_complete") is not False
1072 and _is_fetch_cache_hit_allowed(url, egress_context, _cache)
1073 ):
1074 _text = result.get("content") or ""
1075 if "[... content truncated ...]" not in _text:
1076 try:
1077 if _cache.get(url) is None:
1078 _cache.put(
1079 url,
1080 _text,
1081 title=result.get("title"),
1082 url_out=result.get("url") or url,
1083 )
1084 except Exception: # allow: silent-exception
1085 pass
1087 return result, url, None, current_count
1090def _make_full_fetch_tool(
1091 collector: Any,
1092 settings_snapshot: dict | None = None,
1093 egress_context: Any = None,
1094 library_resolver: Any = None,
1095 max_fetches: int | None = None,
1096 subagent_id: str | None = None,
1097 content_cache: Any = None,
1098 content_max_length: int | None = None,
1099 record_fetch: bool = True,
1100 source_metadata: dict | None = None,
1101 url_guard: Any = None,
1102 availability: Any = None,
1103):
1104 mode_label = "full"
1105 limit = (
1106 max_fetches
1107 if max_fetches is not None
1108 else _read_max_fetches_setting(settings_snapshot)
1109 )
1110 desc = (
1111 "Download and read the full text content from a URL. "
1112 "Use when search snippets aren't detailed enough."
1113 )
1114 if limit > 0:
1115 desc += f" (Up to {limit} page fetches allowed per topic.)"
1117 @tool("fetch_content", description=desc)
1118 def fetch_content(url: str) -> str:
1119 """Download and read the full text content from a URL. Use when search snippets aren't detailed enough."""
1120 from local_deep_research.security.egress.policy import (
1121 PolicyDeniedError,
1122 )
1124 archive_metadata = source_metadata
1125 try:
1126 archive_metadata, archive_guard = _archive_fetch_context(
1127 url, collector, source_metadata, url_guard
1128 )
1129 result, url, err_msg, current_count = _fetch_raw_content(
1130 url,
1131 library_resolver,
1132 collector,
1133 egress_context,
1134 settings_snapshot,
1135 mode_label,
1136 max_fetches=limit if record_fetch else None,
1137 subagent_id=subagent_id,
1138 content_cache=content_cache,
1139 content_max_length=content_max_length,
1140 url_guard=archive_guard,
1141 source_metadata=archive_metadata,
1142 availability=availability,
1143 )
1144 if err_msg is not None:
1145 return err_msg
1147 if result.get("status") == "success":
1148 title = result.get("title", "")
1149 content = result.get("content", "")
1150 if archive_metadata is not None and not (content or "").strip(): 1150 ↛ 1151line 1150 didn't jump to line 1151 because the condition on line 1150 was never true
1151 return f"NOT RELEVANT (no extractable content at {url})"
1152 cite_idx = _register_in_collector(
1153 collector, url, title, content, archive_metadata
1154 )
1155 msg = _format_fetch_message(
1156 cite_idx, title, url, content, archive_metadata
1157 )
1158 if limit > 0 and current_count is not None:
1159 msg += f"\n\n({current_count} of {limit} page fetches used)"
1160 return msg
1161 # result['error'] comes from ContentFetcher, which returns a
1162 # raw str(exception) — scrub it (and the url) before this
1163 # reaches the agent/LLM and user-visible output (#4633).
1164 return sanitize_error_for_agent(
1165 f"Failed to fetch {url}: {result.get('error', 'unknown error')}"
1166 )
1167 except PolicyDeniedError as exc:
1168 # An out-of-scope URL is a RECOVERABLE, per-call decision (the agent
1169 # picked one bad URL among many). Return it as a tool message — like
1170 # the transient-error path below — so the lead agent and pooled
1171 # subagents handle it identically and the agent can adapt, instead
1172 # of re-raising (which aborts a subagent and depends on each agent's
1173 # tool-error layer). The URL was already NOT fetched; the policy
1174 # already enforced — only the REPORTING changes, not security.
1175 target_url = getattr(exc, "target", "") or url
1176 return sanitize_error_for_agent(
1177 f"Cannot fetch {target_url}: blocked by egress policy "
1178 f"({_denial_reason(exc)}). {_EGRESS_DENIAL_HINT}"
1179 )
1180 except Exception as exc:
1181 archive_failure = _archive_failure_message(
1182 exc, availability, archive_metadata
1183 )
1184 if archive_failure is not None:
1185 return archive_failure
1186 # logger.opt(exception=False).error — explicit no-traceback form
1187 # that satisfies both custom-checks (no bare logger.error in an
1188 # except block) and the security contract (no logger.exception
1189 # traceback, which can embed the raw URL). Message carries the mode
1190 # + a REDACTED scheme://host only (no userinfo/path/query); the
1191 # agent/user-facing return is scrubbed separately below.
1192 target_url = getattr(exc, "target", "") or url
1193 logger.opt(exception=False).error(
1194 f"[FETCH] mode={mode_label} status=error "
1195 f"url={redact_url_for_log(target_url)}"
1196 )
1197 return sanitize_error_for_agent(
1198 f"Error fetching {target_url}: {exc}"
1199 )
1201 return fetch_content
1204def _make_summary_fetch_tool(
1205 collector: Any,
1206 model: BaseChatModel,
1207 overall_query: str | None,
1208 settings_snapshot: dict | None = None,
1209 egress_context: Any = None,
1210 library_resolver: Any = None,
1211 max_fetches: int | None = None,
1212 subagent_id: str | None = None,
1213 content_cache: Any = None,
1214 content_max_length: int | None = None,
1215 record_fetch: bool = True,
1216 source_metadata: dict | None = None,
1217 url_guard: Any = None,
1218 availability: Any = None,
1219):
1220 """Build the summary-mode fetch tool.
1222 overall_query=None → focus-only prompt (``summary_focus`` mode).
1223 overall_query=str → focus + overall-query prompt (``summary_focus_query``).
1224 """
1225 use_query = bool(overall_query)
1226 template = SUMMARY_FOCUS_QUERY_PROMPT if use_query else SUMMARY_FOCUS_PROMPT
1228 mode_label = "summary_focus_query" if use_query else "summary_focus"
1229 limit = (
1230 max_fetches
1231 if max_fetches is not None
1232 else _read_max_fetches_setting(settings_snapshot)
1233 )
1234 desc = (
1235 "Fetch a URL and return only the spans of text relevant to `focus`. "
1236 "Pass the specific question or claim you want answered as `focus` — "
1237 "the tool will quote relevant facts verbatim and discard unrelated content."
1238 )
1239 if limit > 0:
1240 desc += f" (Up to {limit} page fetches allowed per topic.)"
1242 @tool("fetch_content", description=desc)
1243 def fetch_content(url: str, focus: str) -> str:
1244 """Fetch a URL and return only the spans of text relevant to ``focus``.
1245 Pass the specific question or claim you want answered as ``focus`` —
1246 the tool will quote relevant facts verbatim and discard unrelated content.
1247 """
1248 from local_deep_research.security.egress.policy import (
1249 PolicyDeniedError,
1250 )
1252 archive_metadata = source_metadata
1253 try:
1254 prompt_char_limit = None
1255 invocation_kwargs = {}
1256 if content_max_length is not None:
1257 page_budget = content_max_length
1258 else:
1259 # Validated first, so an invalid budget fails before the
1260 # window lookup and any fetch.
1261 page_budget = _read_summary_input_max_length_setting(
1262 settings_snapshot
1263 )
1264 prompt_char_limit, invocation_kwargs = (
1265 _summary_invocation_budget(model, settings_snapshot)
1266 )
1267 if prompt_char_limit is not None:
1268 page_budget = min(page_budget, prompt_char_limit)
1269 archive_metadata, archive_guard = _archive_fetch_context(
1270 url, collector, source_metadata, url_guard
1271 )
1272 result, url, err_msg, current_count = _fetch_raw_content(
1273 url,
1274 library_resolver,
1275 collector,
1276 egress_context,
1277 settings_snapshot,
1278 mode_label,
1279 max_fetches=limit if record_fetch else None,
1280 subagent_id=subagent_id,
1281 content_cache=content_cache,
1282 content_max_length=page_budget,
1283 url_guard=archive_guard,
1284 source_metadata=archive_metadata,
1285 availability=availability,
1286 )
1287 if err_msg is not None:
1288 return err_msg
1290 if result.get("status") != "success":
1291 # result['error'] comes from ContentFetcher, which returns a
1292 # raw str(exception) — scrub it (and the url) before this
1293 # reaches the agent/LLM / user output (#4633).
1294 return sanitize_error_for_agent(
1295 f"Failed to fetch {url}: "
1296 f"{result.get('error', 'unknown error')}"
1297 )
1299 title = result.get("title") or ""
1300 content = result.get("content") or ""
1302 # Guard 1 — empty page content (paywalls, JS-only SPAs that
1303 # static fetch can't render, deleted pages with HTTP 200).
1304 # Skipping the LLM call here means we don't pay the round-trip
1305 # to summarise nothing, AND we don't register an empty
1306 # citation in the collector — `_register_in_collector` caches
1307 # by URL, so an empty snippet would lock the URL in as
1308 # "already fetched, nothing here" and the agent would never
1309 # retry it under a different focus.
1310 if not content.strip():
1311 logger.info(
1312 f"[FETCH] mode={mode_label} status=empty_content "
1313 f"url={redact_url_for_log(url)}"
1314 )
1315 return f"NOT RELEVANT (no extractable content at {url})"
1317 fmt_kwargs = {
1318 "focus": focus,
1319 "title": title,
1320 "url": url,
1321 "content": content,
1322 }
1323 if use_query:
1324 fmt_kwargs["overall_query"] = overall_query
1325 prompt = template.format(**fmt_kwargs)
1326 if (
1327 prompt_char_limit is not None
1328 and len(prompt) > prompt_char_limit
1329 ):
1330 # Cut the page tail so the whole prompt fits. At least one
1331 # character is kept, so a window too small for the prompt's
1332 # own text still overflows.
1333 excess = len(prompt) - prompt_char_limit
1334 content = content[: max(1, len(content) - excess)]
1335 fmt_kwargs["content"] = content
1336 prompt = template.format(**fmt_kwargs)
1338 try:
1339 summary_msg = invoke_llm_sync(
1340 model, prompt, **invocation_kwargs
1341 )
1342 summary = getattr(
1343 summary_msg, "content", str(summary_msg)
1344 ).strip()
1345 except Exception as exc:
1346 # Explicit no-traceback form (see full-fetch variant): a
1347 # redacted scheme://host + mode only, no page content, focus,
1348 # or credentials. Agent-facing return is scrubbed below.
1349 logger.opt(exception=False).error(
1350 f"[FETCH] mode={mode_label} status=summary_error "
1351 f"url={redact_url_for_log(url)}"
1352 )
1353 return sanitize_error_for_agent(
1354 f"Error summarizing {url}: {exc}"
1355 )
1357 # Guard 2 — empty LLM summary. The model decided nothing on
1358 # the page answers the focus (or it returned a malformed/empty
1359 # response). Treat as NOT RELEVANT and skip collector
1360 # registration: the agent should be free to re-fetch the URL
1361 # later with a different focus instead of seeing it as
1362 # already-cached with an empty body.
1363 if not summary:
1364 logger.info(
1365 f"[FETCH] mode={mode_label} status=empty_summary "
1366 f"url={redact_url_for_log(url)} "
1367 f"page_chars={len(content)}"
1368 )
1369 return f"NOT RELEVANT (no spans matched focus at {url})"
1371 # Bounded success event: mode, status, redacted origin, and
1372 # page/summary character counts only. Raw focus, overall query,
1373 # title, body, and summary text are deliberately NOT logged —
1374 # they persist to DB/frontend sinks and, for private
1375 # collection/library documents, are sensitive.
1376 logger.info(
1377 f"[FETCH] mode={mode_label} status=success "
1378 f"url={redact_url_for_log(url)} "
1379 f"page_chars={len(content)} summary_chars={len(summary)}"
1380 )
1382 cite_idx = _register_in_collector(
1383 collector, url, title, summary, archive_metadata
1384 )
1385 msg = _format_fetch_message(
1386 cite_idx, title, url, summary, archive_metadata
1387 )
1388 if limit > 0 and current_count is not None:
1389 msg += f"\n\n({current_count} of {limit} page fetches used)"
1390 return msg
1391 except PolicyDeniedError as exc:
1392 # Recoverable per-URL denial — return a tool message so both the
1393 # lead agent and pooled subagents handle it identically and the
1394 # agent stays in-scope. The URL was already NOT fetched; only the
1395 # reporting changes, not security. (See the full-fetch variant.)
1396 target_url = getattr(exc, "target", "") or url
1397 return sanitize_error_for_agent(
1398 f"Cannot fetch {target_url}: blocked by egress policy "
1399 f"({_denial_reason(exc)}). {_EGRESS_DENIAL_HINT}"
1400 )
1401 except Exception as exc:
1402 archive_failure = _archive_failure_message(
1403 exc, availability, archive_metadata
1404 )
1405 if archive_failure is not None:
1406 return archive_failure
1407 # logger.opt(exception=False).error — explicit no-traceback form
1408 # that satisfies both custom-checks (no bare logger.error in an
1409 # except block) and the security contract (no logger.exception
1410 # traceback, which can embed the raw URL). Message carries the mode
1411 # + a REDACTED scheme://host only (no userinfo/path/query); the
1412 # agent/user-facing return is scrubbed separately below.
1413 target_url = getattr(exc, "target", "") or url
1414 logger.opt(exception=False).error(
1415 f"[FETCH] mode={mode_label} status=error "
1416 f"url={redact_url_for_log(target_url)}"
1417 )
1418 return sanitize_error_for_agent(
1419 f"Error fetching {target_url}: {exc}"
1420 )
1422 return fetch_content
1425def build_fetch_tool(
1426 mode: str,
1427 collector: Any,
1428 *,
1429 model: BaseChatModel | None = None,
1430 overall_query: str = "",
1431 settings_snapshot: dict | None = None,
1432 egress_context: Any = None,
1433 library_resolver: Any = None,
1434 max_fetches: int | None = None,
1435 subagent_id: str | None = None,
1436 content_cache: Any = None,
1437 content_max_length: int | None = None,
1438 record_fetch: bool = True,
1439 source_metadata: dict | None = None,
1440 url_guard: Any = None,
1441 availability: Any = None,
1442):
1443 """Build the agent-facing ``fetch_content`` tool for *mode*.
1445 ``record_fetch=False`` is used by the archive wrapper after it reserves
1446 the shared fetch slot before archive lookup. ``source_metadata`` carries
1447 that capture's provenance into the citation collector.
1449 Returns ``None`` when ``mode == 'disabled'``; the caller should not
1450 register the tool with the agent in that case (and the system prompt
1451 should also drop the corresponding instruction line so the agent
1452 isn't told to use a tool that doesn't exist).
1454 ``settings_snapshot`` is captured by the tool closure so the per-call
1455 JS-rendering toggle can be read on a worker thread (where
1456 ``threading.local`` context does not propagate).
1458 ``content_max_length`` explicitly overrides the page-text budget in
1459 any mode, without a context-window limit. Otherwise, ``full`` reads
1460 ``search.fetch.content_max_length`` and summary modes read
1461 ``search.fetch.summary_input_max_length``. Unless *model* was built
1462 for a cloud provider with unrestricted context, summary modes then cut
1463 the page text so the whole extraction prompt fits the model's context
1464 window minus its response allowance, estimated at four characters per
1465 token (see ``_summary_invocation_budget``). That budget limits only the
1466 separate extraction prompt, not the tool's summary returned to the
1467 research agent.
1469 ``egress_context`` is captured by the closure so the per-call URL
1470 can be policy-gated; when ``None``, no policy enforcement runs
1471 (preserves legacy non-LangGraph callers).
1473 ``library_resolver`` is captured by the closure so a fetch call can
1474 short-circuit on ``/library/document/<uuid>[/pdf]`` URLs (a local DB
1475 read, not a network fetch) and on bare ``[N]`` citation markers
1476 (rewritten to the citation's URL via ``SearchResultsCollector``).
1477 When ``None``, the tool falls through to the egress gate unchanged
1478 (which, when ``egress_context`` is configured, rejects library / citation
1479 URLs as ``unsupported_scheme``) — same as the pre-fix behaviour.
1481 ``content_cache`` is the single shared cache for the whole research
1482 run. When None, the cache bound to the current research (if any) is
1483 used, so the lead agent, subagents, quick phase and subsections all
1484 share previously fetched pages instead of re-fetching on the spot.
1485 """
1486 if mode == "disabled":
1487 return None
1488 if mode == "full":
1489 return _make_full_fetch_tool(
1490 collector,
1491 settings_snapshot=settings_snapshot,
1492 egress_context=egress_context,
1493 library_resolver=library_resolver,
1494 max_fetches=max_fetches,
1495 subagent_id=subagent_id,
1496 content_cache=content_cache,
1497 content_max_length=content_max_length,
1498 record_fetch=record_fetch,
1499 source_metadata=source_metadata,
1500 url_guard=url_guard,
1501 availability=availability,
1502 )
1503 if mode == "summary_focus":
1504 if model is None:
1505 raise ValueError("summary_focus fetch mode requires a model")
1506 return _make_summary_fetch_tool(
1507 collector,
1508 model,
1509 overall_query=None,
1510 settings_snapshot=settings_snapshot,
1511 egress_context=egress_context,
1512 library_resolver=library_resolver,
1513 max_fetches=max_fetches,
1514 subagent_id=subagent_id,
1515 content_cache=content_cache,
1516 content_max_length=content_max_length,
1517 record_fetch=record_fetch,
1518 source_metadata=source_metadata,
1519 url_guard=url_guard,
1520 availability=availability,
1521 )
1522 if mode == "summary_focus_query":
1523 if model is None:
1524 raise ValueError("summary_focus_query fetch mode requires a model")
1525 # Empty overall_query falls back to focus-only behaviour at format
1526 # time; we keep the *_query mode label so logs stay diagnostic.
1527 return _make_summary_fetch_tool(
1528 collector,
1529 model,
1530 overall_query=overall_query or None,
1531 settings_snapshot=settings_snapshot,
1532 egress_context=egress_context,
1533 library_resolver=library_resolver,
1534 max_fetches=max_fetches,
1535 subagent_id=subagent_id,
1536 content_cache=content_cache,
1537 content_max_length=content_max_length,
1538 record_fetch=record_fetch,
1539 source_metadata=source_metadata,
1540 url_guard=url_guard,
1541 availability=availability,
1542 )
1543 raise ValueError(
1544 f"Unknown fetch mode {mode!r}; expected one of {FETCH_MODES}"
1545 )
1548__all__ = [
1549 "DEFAULT_MAX_FETCHES_PER_TOPIC",
1550 "FETCH_MODES",
1551 "build_fetch_tool",
1552 "make_library_resolver",
1553]