Coverage for src/local_deep_research/advanced_search_system/tools/fetch/__init__.py: 97%
176 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +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_client,
41)
43from .library_resolver import (
44 is_citation_reference,
45 make_library_resolver,
46 resolve_citation_reference as _resolve_citation_reference,
47)
48from .prompts import SUMMARY_FOCUS_PROMPT, SUMMARY_FOCUS_QUERY_PROMPT
51# Per-call timeouts and caps. Kept here rather than in the strategy file
52# because they are properties of the fetch tool, not of agent
53# orchestration.
54CONTENT_FETCH_TIMEOUT = 30
55CONTENT_MAX_LENGTH = 10_000
57# Cap for credential-scrubbed fetch-tool error strings. Larger than the
58# 200-char HTTP-client default because these errors feed the agent's
59# reasoning; credential scrubbing still runs first on the full string (#4633).
60_TOOL_ERROR_MAX_LEN = 500
63def _scrub_tool_error(message: str) -> str:
64 """Scrub credentials from an LLM/agent-facing fetch-tool error string."""
65 return sanitize_error_for_client(message, max_length=_TOOL_ERROR_MAX_LEN)
68FETCH_MODES = (
69 "disabled",
70 "full",
71 "summary_focus",
72 "summary_focus_query",
73)
76def _register_in_collector(
77 collector: Any,
78 url: str,
79 title: str,
80 snippet_source: str,
81) -> int:
82 """Register a fetched URL in the collector and return its 1-based citation index.
84 If the URL was already tracked (via a prior search hit) the existing
85 index is reused so the agent sees a stable citation per URL.
87 .. note::
88 The ``find_or_add_result``-less branch below is **speculative
89 generality**: ``SearchResultsCollector`` is the only collector in
90 this repo and it has that method, so nothing in-tree reaches the
91 fallback and no test exercises it against a real collector. It
92 exists for third-party or pre-#5363 collectors. Treat it as
93 unverified against any live caller — do not assume a bug there is
94 affecting production, and do not grow it further without a real
95 consumer to justify it.
97 This branch was left at its pre-existing shape deliberately. An
98 earlier revision of this follow-up grew it — duck-typed tuple
99 unpacking for a bare-``int`` ``add_results`` contract, a ``getattr``
100 guard on ``find_by_url`` — for a collector that does not exist in
101 any repo we can see, which is precisely what the paragraph above
102 forbids. That growth was reverted rather than shipped.
103 """
104 snippet = snippet_source[:200].strip()
105 if len(snippet_source) > 200:
106 snippet += "..."
107 if not hasattr(collector, "find_or_add_result"):
108 # Legacy/custom collectors without find_or_add_result fallback
109 _, indexed = collector.add_results(
110 [{"title": title, "link": url, "snippet": snippet}],
111 engine_name="fetch",
112 )
113 if indexed: 113 ↛ 123line 113 didn't jump to line 123 because the condition on line 113 was always true
114 raw_idx = indexed[0].get("index")
115 try:
116 return int(raw_idx)
117 except (ValueError, TypeError):
118 logger.warning(
119 "Failed to parse citation index for url={}: {!r}",
120 redact_url_for_log(url),
121 raw_idx,
122 )
123 existing_idx = collector.find_by_url(url)
124 if existing_idx is not None:
125 return existing_idx
126 raise RuntimeError(
127 f"Failed to register fetched URL in collector: {redact_url_for_log(url)}"
128 )
130 return collector.find_or_add_result(
131 {"title": title, "link": url, "snippet": snippet},
132 engine_name="fetch",
133 )
136def _enforce_url_policy(url: str, egress_context: Any) -> None:
137 """Run ``evaluate_url`` against ``egress_context`` and raise
138 ``PolicyDeniedError`` on denial.
140 No-op when no context is configured (callers without policy enforcement,
141 e.g. legacy non-LangGraph strategies, see the legacy behavior).
142 """
143 if egress_context is None:
144 return
145 from local_deep_research.security.egress.policy import (
146 PolicyDeniedError,
147 evaluate_url,
148 )
150 decision = evaluate_url(url, egress_context)
151 if not decision.allowed:
152 raise PolicyDeniedError(decision, target=url)
155def _denial_reason(exc: Any) -> str:
156 """Best-effort egress-denial reason code for an agent-facing message."""
157 return getattr(getattr(exc, "decision", None), "reason", "policy_denied")
160# Shared instruction appended to every per-URL egress denial returned to the
161# agent. Tells it WHY the fetch was refused and what to do instead, so it
162# adapts (stays in-scope) rather than retrying the same out-of-scope URL.
163_EGRESS_DENIAL_HINT = (
164 "In this run only local collection/library documents can be fetched; "
165 "skip external URLs."
166)
168# ---------------------------------------------------------------------------
169# Pre-resolution: rewrite non-network URLs BEFORE the egress gate.
170#
171# Two URL-shaped strings reach the fetch tool that aren't actually network
172# fetches and that the egress policy correctly rejects as
173# ``unsupported_scheme``. Resolving them here lets the agent get the page
174# content (or a helpful error) instead of a generic denial — A3 from
175# research_f3045c5b_issue_analysis.md.
176# ---------------------------------------------------------------------------
178# ``_KIND_RESULT`` marks a pre-resolved local document; the fetch tool
179# consumes the dict directly. ``_KIND_ERROR`` short-circuits the fetch
180# tool with the message in ``payload`` (already routed through the
181# credential scrubber by the caller). ``_KIND_REWRITTEN`` means "this
182# URL was rewritten to a new one — use the new URL going forward but
183# keep running the normal HTTP flow" (e.g. a ``[N]`` citation marker
184# that resolves to an external URL). ``None`` means "no pre-resolution;
185# run the normal HTTP path with the original URL."
186_KIND_RESULT = "result"
187_KIND_ERROR = "error"
188_KIND_REWRITTEN = "rewritten"
191def _try_resolve_url(
192 url: str,
193 library_resolver: Any,
194 collector: Any,
195 _visited: set[str] | None = None,
196) -> tuple[str, Any] | None:
197 """Pre-resolve a fetch URL that isn't a network URL.
199 Returns one of:
201 - ``(``_KIND_RESULT``, dict)`` — local-content payload shaped like a
202 ``ContentFetcher.fetch()`` success result, ready for the post-fetch
203 pipeline. The caller MUST skip the egress gate (it's a local read).
204 - ``(``_KIND_ERROR``, str)`` — short-circuit with this error message.
205 - ``(``_KIND_REWRITTEN``, str)`` — *url* was rewritten (e.g. a
206 citation marker ``[N]`` that resolved to an external URL); use
207 the new URL but otherwise run the normal HTTP flow (the egress
208 gate still applies).
209 - ``None`` — no pre-resolution; run the normal HTTP path with *url*.
211 Handles:
213 - ``[N]`` citation markers → ``SearchResultsCollector.find_by_index``,
214 then recurse with the resolved URL (so a citation whose source is a
215 library doc still hits the local fast path, while a citation whose
216 source is external surfaces a ``_KIND_REWRITTEN`` URL for the gate).
217 - ``/library/document/<uuid>[/pdf]`` → ``Document.text_content`` read
218 from the user DB via the library resolver.
219 """
220 if not isinstance(url, str):
221 return None
223 if _visited is None:
224 _visited = set()
225 if url in _visited:
226 return (
227 _KIND_ERROR,
228 f"Circular citation reference detected for '{url}'.",
229 )
230 if len(_visited) >= 5:
231 return (
232 _KIND_ERROR,
233 f"Citation resolution depth limit exceeded for '{url}'.",
234 )
235 _visited.add(url)
237 # 1. Citation marker [N] → resolve to citation's URL.
238 citation = _resolve_citation_reference(url, collector)
239 if citation is not None:
240 resolved_url = citation.get("link") or citation.get("url") or ""
241 if not resolved_url:
242 return (
243 _KIND_ERROR,
244 (
245 f"Citation {url} has no URL field. "
246 "Use the citation's URL or a different source instead."
247 ),
248 )
249 # First, check if the citation's URL is itself a library doc
250 # (the recursive call sees the library path and returns _KIND_RESULT
251 # directly). Otherwise surface the citation's URL as a rewrite so
252 # the fetch tool updates its ``url`` variable before the egress
253 # gate / HTTP fetch runs against the resolved URL — not the marker.
254 recursive = _try_resolve_url(
255 resolved_url, library_resolver, collector, _visited=_visited
256 )
257 if recursive is not None:
258 return recursive
259 return (_KIND_REWRITTEN, resolved_url)
261 # 2. Citation marker that doesn't match a tracked citation.
262 # ``_resolve_citation_reference`` only returns the citation for a
263 # well-formed ``[N]`` marker; anything else falls through to the
264 # library-resolver check below.
265 if is_citation_reference(url) is not None:
266 return (
267 _KIND_ERROR,
268 (
269 f"No registered citation matches {url}. The agent's "
270 "search results use citation markers; use the source URL "
271 "(the link next to the marker) or a tracked citation "
272 "instead of a raw marker."
273 ),
274 )
276 # 3. Library document URL.
277 if library_resolver is not None:
278 content = library_resolver(url)
279 if content is not None:
280 text = content.get("content") or ""
281 if len(text) > CONTENT_MAX_LENGTH:
282 text = text[:CONTENT_MAX_LENGTH]
283 return (
284 _KIND_RESULT,
285 {
286 "status": "success",
287 "title": content.get("title") or "",
288 "content": text,
289 "url": content.get("url") or url,
290 },
291 )
293 return None
296def _fetch_raw_content(
297 url: str,
298 library_resolver: Any,
299 collector: Any,
300 egress_context: Any,
301 settings_snapshot: dict | None,
302 mode_label: str,
303) -> tuple[dict | None, str, str | None]:
304 """Pre-resolve non-network URL shapes or fetch via HTTP.
306 Returns:
307 (result, final_url, error_string)
308 If error_string is not None, a pre-resolution error occurred (already
309 scrubbed) and should be returned immediately by the fetch tool.
310 """
311 from local_deep_research.content_fetcher import ContentFetcher
313 pre = _try_resolve_url(url, library_resolver, collector)
314 result: dict | None = None
315 if pre is not None:
316 kind, payload = pre
317 if kind == _KIND_ERROR:
318 return None, url, _scrub_tool_error(payload)
319 if kind == _KIND_RESULT:
320 # A local library document was resolved directly from the user DB.
321 # Skip both the egress gate (local read) and ContentFetcher (no HTTP).
322 result = payload
323 url = result.get("url") or url
324 logger.info(
325 f"[FETCH] mode={mode_label} source=library url={url} — "
326 "resolved local library document directly"
327 )
328 else: # _KIND_REWRITTEN — citation marker resolved to a URL
329 url = payload
331 if result is None:
332 # Either no pre-resolution or a citation-marker rewrite —
333 # run the normal HTTP path. Per-URL egress gate (pre-fetch)
334 # + ContentFetcher's own per-redirect gate both raise
335 # PolicyDeniedError on an out-of-scope URL. Run the gate
336 # INSIDE the try so the denial is returned as a recoverable
337 # tool message.
338 _enforce_url_policy(url, egress_context)
339 enable_js = _read_js_rendering_setting(settings_snapshot)
340 with ContentFetcher(
341 timeout=CONTENT_FETCH_TIMEOUT,
342 enable_js_rendering=enable_js,
343 egress_context=egress_context,
344 ) as fetcher:
345 result = fetcher.fetch(url, max_length=CONTENT_MAX_LENGTH)
347 return result, url, None
350def _make_full_fetch_tool(
351 collector: Any,
352 settings_snapshot: dict | None = None,
353 egress_context: Any = None,
354 library_resolver: Any = None,
355):
356 mode_label = "full"
358 @tool
359 def fetch_content(url: str) -> str:
360 """Download and read the full text content from a URL. Use when search snippets aren't detailed enough."""
361 from local_deep_research.security.egress.policy import (
362 PolicyDeniedError,
363 )
365 try:
366 result, url, err_msg = _fetch_raw_content(
367 url,
368 library_resolver,
369 collector,
370 egress_context,
371 settings_snapshot,
372 mode_label,
373 )
374 if err_msg is not None:
375 return err_msg
377 if result.get("status") == "success":
378 title = result.get("title", "")
379 content = result.get("content", "")
380 cite_idx = _register_in_collector(
381 collector, url, title, content
382 )
383 return f"[{cite_idx}] Title: {title}\nURL: {url}\n\n{content}"
384 # result['error'] comes from ContentFetcher, which returns a
385 # raw str(exception) — scrub it (and the url) before this
386 # reaches the agent/LLM and user-visible output (#4633).
387 return _scrub_tool_error(
388 f"Failed to fetch {url}: {result.get('error', 'unknown error')}"
389 )
390 except PolicyDeniedError as exc:
391 # An out-of-scope URL is a RECOVERABLE, per-call decision (the agent
392 # picked one bad URL among many). Return it as a tool message — like
393 # the transient-error path below — so the lead agent and pooled
394 # subagents handle it identically and the agent can adapt, instead
395 # of re-raising (which aborts a subagent and depends on each agent's
396 # tool-error layer). The URL was already NOT fetched; the policy
397 # already enforced — only the REPORTING changes, not security.
398 target_url = getattr(exc, "target", "") or url
399 return _scrub_tool_error(
400 f"Cannot fetch {target_url}: blocked by egress policy "
401 f"({_denial_reason(exc)}). {_EGRESS_DENIAL_HINT}"
402 )
403 except Exception as exc:
404 # Message carries the mode + a REDACTED scheme://host only (no
405 # userinfo/path/query) so an operator can locate the failure without
406 # the log line leaking credentials, query tokens, or page content.
407 # The traceback follows the sink's diagnose setting (off by default;
408 # see utilities/log_utils). The agent/user-facing return is scrubbed
409 # separately below.
410 target_url = getattr(exc, "target", "") or url
411 logger.exception(
412 "fetch_content tool error (mode={}, url={})",
413 mode_label,
414 redact_url_for_log(target_url),
415 )
416 return _scrub_tool_error(f"Error fetching {target_url}: {exc}")
418 return fetch_content
421def _make_summary_fetch_tool(
422 collector: Any,
423 model: BaseChatModel,
424 overall_query: str | None,
425 settings_snapshot: dict | None = None,
426 egress_context: Any = None,
427 library_resolver: Any = None,
428):
429 """Build the summary-mode fetch tool.
431 overall_query=None → focus-only prompt (``summary_focus`` mode).
432 overall_query=str → focus + overall-query prompt (``summary_focus_query``).
433 """
434 use_query = bool(overall_query)
435 template = SUMMARY_FOCUS_QUERY_PROMPT if use_query else SUMMARY_FOCUS_PROMPT
437 mode_label = "summary_focus_query" if use_query else "summary_focus"
439 @tool
440 def fetch_content(url: str, focus: str) -> str:
441 """Fetch a URL and return only the spans of text relevant to ``focus``.
442 Pass the specific question or claim you want answered as ``focus`` —
443 the tool will quote relevant facts verbatim and discard unrelated content.
444 """
445 from local_deep_research.security.egress.policy import (
446 PolicyDeniedError,
447 )
449 try:
450 result, url, err_msg = _fetch_raw_content(
451 url,
452 library_resolver,
453 collector,
454 egress_context,
455 settings_snapshot,
456 mode_label,
457 )
458 if err_msg is not None:
459 return err_msg
461 if result.get("status") != "success": 461 ↛ 465line 461 didn't jump to line 465 because the condition on line 461 was never true
462 # result['error'] comes from ContentFetcher, which returns a
463 # raw str(exception) — scrub it (and the url) before this
464 # reaches the agent/LLM / user output (#4633).
465 return _scrub_tool_error(
466 f"Failed to fetch {url}: "
467 f"{result.get('error', 'unknown error')}"
468 )
470 title = result.get("title") or ""
471 content = result.get("content") or ""
473 # Guard 1 — empty page content (paywalls, JS-only SPAs that
474 # static fetch can't render, deleted pages with HTTP 200).
475 # Skipping the LLM call here means we don't pay the round-trip
476 # to summarise nothing, AND we don't register an empty
477 # citation in the collector — `_register_in_collector` caches
478 # by URL, so an empty snippet would lock the URL in as
479 # "already fetched, nothing here" and the agent would never
480 # retry it under a different focus.
481 if not content.strip():
482 logger.info(
483 f"[FETCH] mode={mode_label} url={url} — "
484 "empty page content, returning NOT RELEVANT without "
485 "LLM call or collector registration"
486 )
487 return f"NOT RELEVANT (no extractable content at {url})"
489 fmt_kwargs = {
490 "focus": focus,
491 "title": title,
492 "url": url,
493 "content": content,
494 }
495 if use_query:
496 fmt_kwargs["overall_query"] = overall_query
497 prompt = template.format(**fmt_kwargs)
499 try:
500 summary_msg = model.invoke(prompt)
501 summary = getattr(
502 summary_msg, "content", str(summary_msg)
503 ).strip()
504 except Exception as exc:
505 # Redacted scheme://host + mode only — no page content,
506 # focus, or credentials in the message. Traceback follows
507 # the sink's diagnose setting (off by default).
508 logger.exception(
509 "fetch_content summary LLM error (mode={}, url={})",
510 mode_label,
511 redact_url_for_log(url),
512 )
513 return _scrub_tool_error(f"Error summarizing {url}: {exc}")
515 # Diagnostic log: per-fetch input/output for evaluating the
516 # summariser. Single multi-line block so it's atomic per call
517 # and easy to grep with ``grep -A1000 "[FETCH] mode="``.
518 log_lines = [
519 f"[FETCH] mode={mode_label} url={url}",
520 f"[FETCH] focus: {focus}",
521 ]
522 if use_query:
523 log_lines.append(f"[FETCH] overall_query: {overall_query}")
524 log_lines.extend(
525 [
526 f"[FETCH] title: {title}",
527 f"[FETCH] page_text ({len(content)} chars):",
528 content,
529 f"[FETCH] summary returned ({len(summary)} chars):",
530 summary or "(empty)",
531 "[FETCH] ---",
532 ]
533 )
534 logger.info("\n".join(log_lines))
536 # Guard 2 — empty LLM summary. The model decided nothing on
537 # the page answers the focus (or it returned a malformed/empty
538 # response). Treat as NOT RELEVANT and skip collector
539 # registration: the agent should be free to re-fetch the URL
540 # later with a different focus instead of seeing it as
541 # already-cached with an empty body.
542 if not summary:
543 return f"NOT RELEVANT (no spans matched focus at {url})"
545 cite_idx = _register_in_collector(collector, url, title, summary)
546 return f"[{cite_idx}] Title: {title}\nURL: {url}\n\n{summary}"
547 except PolicyDeniedError as exc:
548 # Recoverable per-URL denial — return a tool message so both the
549 # lead agent and pooled subagents handle it identically and the
550 # agent stays in-scope. The URL was already NOT fetched; only the
551 # reporting changes, not security. (See the full-fetch variant.)
552 target_url = getattr(exc, "target", "") or url
553 return _scrub_tool_error(
554 f"Cannot fetch {target_url}: blocked by egress policy "
555 f"({_denial_reason(exc)}). {_EGRESS_DENIAL_HINT}"
556 )
557 except Exception as exc:
558 # Message carries the mode + a REDACTED scheme://host only (no
559 # userinfo/path/query) so an operator can locate the failure without
560 # the log line leaking credentials, query tokens, or page content.
561 # The traceback follows the sink's diagnose setting (off by default;
562 # see utilities/log_utils). The agent/user-facing return is scrubbed
563 # separately below.
564 target_url = getattr(exc, "target", "") or url
565 logger.exception(
566 "fetch_content tool error (mode={}, url={})",
567 mode_label,
568 redact_url_for_log(target_url),
569 )
570 return _scrub_tool_error(f"Error fetching {target_url}: {exc}")
572 return fetch_content
575def build_fetch_tool(
576 mode: str,
577 collector: Any,
578 *,
579 model: BaseChatModel | None = None,
580 overall_query: str = "",
581 settings_snapshot: dict | None = None,
582 egress_context: Any = None,
583 library_resolver: Any = None,
584):
585 """Build the agent-facing ``fetch_content`` tool for *mode*.
587 Returns ``None`` when ``mode == 'disabled'``; the caller should not
588 register the tool with the agent in that case (and the system prompt
589 should also drop the corresponding instruction line so the agent
590 isn't told to use a tool that doesn't exist).
592 ``settings_snapshot`` is captured by the tool closure so the per-call
593 JS-rendering toggle can be read on a worker thread (where
594 ``threading.local`` context does not propagate).
596 ``egress_context`` is captured by the closure so the per-call URL
597 can be policy-gated; when ``None``, no policy enforcement runs
598 (preserves legacy non-LangGraph callers).
600 ``library_resolver`` is captured by the closure so a fetch call can
601 short-circuit on ``/library/document/<uuid>[/pdf]`` URLs (a local DB
602 read, not a network fetch) and on bare ``[N]`` citation markers
603 (rewritten to the citation's URL via ``SearchResultsCollector``).
604 When ``None``, the tool falls through to the egress gate unchanged
605 (which, when ``egress_context`` is configured, rejects library / citation
606 URLs as ``unsupported_scheme``) — same as the pre-fix behaviour.
607 """
608 if mode == "disabled":
609 return None
610 if mode == "full":
611 return _make_full_fetch_tool(
612 collector,
613 settings_snapshot=settings_snapshot,
614 egress_context=egress_context,
615 library_resolver=library_resolver,
616 )
617 if mode == "summary_focus":
618 if model is None:
619 raise ValueError("summary_focus fetch mode requires a model")
620 return _make_summary_fetch_tool(
621 collector,
622 model,
623 overall_query=None,
624 settings_snapshot=settings_snapshot,
625 egress_context=egress_context,
626 library_resolver=library_resolver,
627 )
628 if mode == "summary_focus_query":
629 if model is None:
630 raise ValueError("summary_focus_query fetch mode requires a model")
631 # Empty overall_query falls back to focus-only behaviour at format
632 # time; we keep the *_query mode label so logs stay diagnostic.
633 return _make_summary_fetch_tool(
634 collector,
635 model,
636 overall_query=overall_query or None,
637 settings_snapshot=settings_snapshot,
638 egress_context=egress_context,
639 library_resolver=library_resolver,
640 )
641 raise ValueError(
642 f"Unknown fetch mode {mode!r}; expected one of {FETCH_MODES}"
643 )
646__all__ = ["FETCH_MODES", "build_fetch_tool", "make_library_resolver"]