"""
LangGraph agent-based research strategy with parallel subagent support.

Uses LangChain's create_agent() to build a tool-calling agent that autonomously
decides what to search, when to dig deeper, and when to synthesize. Complex
questions can be decomposed into subtopics researched in parallel by subagents.
"""

from __future__ import annotations

import threading
from concurrent.futures import ThreadPoolExecutor, TimeoutError, as_completed
from datetime import UTC, datetime
from typing import Any, Dict

from langchain_core.language_models import BaseChatModel
from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.errors import GraphRecursionError
from loguru import logger

from ...utilities.thread_context import get_search_context, search_context
from ...citation_handler import CitationHandler
from ..tools.fetch import FETCH_MODES, build_fetch_tool
from .base_strategy import BaseSearchStrategy
from ...security import sanitize_error_for_client

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

DEFAULT_MAX_ITERATIONS = (
    50  # agent needs many more cycles than pipeline strategies
)
MIN_ITERATIONS = 10  # below this the agent can barely do anything useful
SUBAGENT_TIMEOUT_SECONDS = 1800  # 30 minutes per subagent
MAX_SUBTOPICS = 8
MAX_SUBAGENT_WORKERS = 4
# CONTENT_FETCH_TIMEOUT and CONTENT_MAX_LENGTH live alongside the fetch
# tool builders in advanced_search_system/tools/fetch/.

# Cap for credential-scrubbed tool/agent error strings. Larger than the
# 200-char HTTP-client default of ``sanitize_error_for_client`` because these
# strings feed the agent's reasoning AND the ErrorReporter pattern map, where
# over-aggressive truncation drops the categorizable error signal. Credential
# scrubbing still runs first on the full untruncated string (#4633).
_TOOL_ERROR_MAX_LEN = 500


def _scrub_tool_error(message: str) -> str:
    """Scrub credentials from an LLM/agent-facing tool error string."""
    return sanitize_error_for_client(message, max_length=_TOOL_ERROR_MAX_LEN)


# ---------------------------------------------------------------------------
# Thread-safe search result collector
# ---------------------------------------------------------------------------


class SearchResultsCollector:
    """Accumulates search results from the lead agent and subagents.

    Thread-safe: multiple subagent threads may call ``add_results``
    concurrently.  The ``_all_links`` reference points to the strategy's
    shared ``all_links_of_system`` list and is never reassigned.
    """

    def __init__(self, all_links: list | None = None) -> None:
        self._results: list[dict] = []
        self._sources: list[str] = []
        self._lock = threading.Lock()
        self._all_links = all_links if all_links is not None else []

    # -- public API ----------------------------------------------------------

    def add_results(
        self,
        results: list[dict],
        engine_name: str = "web",
    ) -> int:
        """Index *results* and append to the internal list **and** the shared
        ``all_links_of_system``.  Returns the starting citation index
        (0-based) assigned to the first result in this batch.

        The entire operation runs under a single lock acquisition so that
        citation indices are never duplicated.
        """
        if not results:
            return len(self._all_links)

        with self._lock:
            # Use global offset (all_links) not per-call offset (results)
            # so that indices are unique across sections in detailed reports.
            start_idx = len(self._all_links)
            for i, raw in enumerate(results):
                if not isinstance(raw, dict):
                    continue
                r = dict(raw)  # shallow copy to avoid mutating engine output
                r["index"] = str(start_idx + i + 1)
                r["source_engine"] = engine_name
                # Normalise URL key — citation handler expects "link"
                if "link" not in r and "url" in r:
                    r["link"] = r["url"]
                self._results.append(r)
                link = r.get("link", "")
                if link:
                    self._sources.append(link)
                self._all_links.append(r)
            return start_idx

    def find_by_url(self, url: str) -> int | None:
        """Return the 1-based citation index if *url* is already tracked, else ``None``."""
        with self._lock:
            for r in self._all_links:
                if r.get("link", r.get("url", "")) == url:
                    idx = r.get("index")
                    if idx is not None:
                        return int(idx)
                    return None
            return None

    def reset(self) -> None:
        """Clear per-call state.  ``_all_links`` is intentionally kept."""
        with self._lock:
            self._results.clear()
            self._sources.clear()

    @property
    def results(self) -> list[dict]:
        with self._lock:
            return list(self._results)

    @property
    def sources(self) -> list[str]:
        with self._lock:
            return list(self._sources)


# ---------------------------------------------------------------------------
# Tool factory helpers
# ---------------------------------------------------------------------------


# User-facing names for the agent's tools — used in the live milestone
# messages so the chat thinking-text reads "Searching PubMed for …"
# instead of "Tool: search_pubmed — …". Falls back to title-casing the
# raw tool name for tools without an explicit entry, so newly added
# engines work cleanly without a code change.
_TOOL_DISPLAY_NAMES = {
    "web_search": "the web",
    "search_pubmed": "PubMed",
    "search_arxiv": "arXiv",
    "search_semantic_scholar": "Semantic Scholar",
    "search_openalex": "OpenAlex",
    "search_searxng": "the web (SearXNG)",
    "search_google_scholar": "Google Scholar",
    "search_brave": "Brave Search",
    "search_duckduckgo": "DuckDuckGo",
    "search_serper": "Google (Serper)",
    "search_scaleserp": "Google (ScaleSERP)",
    "search_wikipedia": "Wikipedia",
    "search_github": "GitHub",
    "search_stackexchange": "Stack Exchange",
    "search_openlibrary": "Open Library",
    "search_gutenberg": "Project Gutenberg",
    "search_pubchem": "PubChem",
    "search_zenodo": "Zenodo",
    "search_nasa_ads": "NASA ADS",
    "search_local": "your library",
    "fetch_url": "the page",
    "research_subtopic": "subtopic researcher",
}


def _tool_display_name(name: str) -> str:
    """Friendly name for a tool, falling back to a cleaned raw name."""
    if name in _TOOL_DISPLAY_NAMES:
        return _TOOL_DISPLAY_NAMES[name]
    # Strip leading "search_" and title-case for unknown engines.
    cleaned = name[len("search_") :] if name.startswith("search_") else name
    return cleaned.replace("_", " ").title()


def _format_results(results: list[dict], start_idx: int) -> str:
    """Format search results as ``[N] Title (URL)\\nSnippet``."""
    lines = []
    for i, r in enumerate(results):
        if not isinstance(r, dict):
            continue
        idx = start_idx + i + 1
        title = r.get("title", "No title")
        link = r.get("link", r.get("url", ""))
        snippet = r.get("snippet", r.get("body", ""))
        lines.append(f"[{idx}] {title} ({link})\n{snippet}")
    return "\n\n".join(lines) if lines else "No results."


def _make_web_search_tool(
    search_engine_name: str,
    model: BaseChatModel,
    settings_snapshot: dict,
    collector: SearchResultsCollector,
    programmatic_mode: bool = False,
):
    """Create a ``web_search`` tool that instantiates a fresh engine per call."""

    @tool
    def web_search(query: str) -> str:
        """Search the web for current information, facts, or news. Returns search result snippets with source indices."""
        from local_deep_research.utilities.resource_utils import safe_close
        from local_deep_research.web_search_engines.search_engine_factory import (
            create_search_engine,
        )

        engine = create_search_engine(
            engine_name=search_engine_name,
            llm=model,
            settings_snapshot=settings_snapshot,
            programmatic_mode=programmatic_mode,
        )
        if engine is None:
            return f"Failed to create search engine '{search_engine_name}'."
        try:
            results = engine.run(query)
            if not isinstance(results, list) or not results:
                return f"No results found for '{query}'. Try rephrasing."
            start = collector.add_results(
                results, engine_name=search_engine_name
            )
            return _format_results(results, start)
        except Exception as exc:
            logger.exception("web_search tool error")
            # Scrub credentials: a search-engine exception can embed the
            # request URL, which may carry an API key. Full detail is logged
            # server-side above.
            return _scrub_tool_error(f"Search error: {exc}")
        finally:
            safe_close(engine, "web search engine")

    return web_search


# Fetch tool builders (full / summary_focus / summary_focus_query / disabled)
# live in ``advanced_search_system.tools.fetch``; see ``build_fetch_tool``.


def _make_specialized_search_tool(
    engine_name: str,
    description: str,
    model: BaseChatModel,
    settings_snapshot: dict,
    collector: SearchResultsCollector,
    programmatic_mode: bool = False,
):
    """Create a ``search_{engine}`` tool for a specific search engine."""

    @tool
    def specialized_search(query: str) -> str:
        """Search a specialized engine."""  # overridden below
        from local_deep_research.utilities.resource_utils import safe_close
        from local_deep_research.web_search_engines.search_engine_factory import (
            create_search_engine,
        )

        engine = create_search_engine(
            engine_name=engine_name,
            llm=model,
            settings_snapshot=settings_snapshot,
            programmatic_mode=programmatic_mode,
        )
        if engine is None:
            return f"Failed to create {engine_name} engine."
        try:
            results = engine.run(query)
            if not isinstance(results, list) or not results:
                return f"No results from {engine_name} for '{query}'. Try rephrasing."
            start = collector.add_results(results, engine_name=engine_name)
            return _format_results(results, start)
        except Exception as exc:
            logger.exception(f"search_{engine_name} tool error")
            return _scrub_tool_error(f"Search error ({engine_name}): {exc}")
        finally:
            safe_close(engine, f"{engine_name} search engine")

    # Override name and description after decoration
    specialized_search.name = f"search_{engine_name}"
    specialized_search.description = description
    return specialized_search


def _make_research_subtopic_tool(
    search_engine_name: str,
    model: BaseChatModel,
    settings_snapshot: dict,
    collector: SearchResultsCollector,
    max_sub_iterations: int,
    progress_callback=None,
    programmatic_mode: bool = False,
    fetch_mode: str = "summary_focus_query",
    overall_query: str = "",
    egress_context=None,
):
    """Create the ``research_subtopic`` tool that spawns parallel subagents.

    ``overall_query`` is the original user query passed by the lead agent's
    strategy; it's forwarded to summary-mode fetch tools so the per-page
    extractor sees both the agent's per-fetch focus and the original
    research question.
    """

    @tool
    def research_subtopic(subtopics: list[str]) -> str:
        """Delegate parallel research on multiple subtopics. Each subtopic is
        investigated by a separate agent. Pass 2-5 focused research questions."""
        from langchain.agents import create_agent

        if not subtopics:
            return "No subtopics provided."
        if len(subtopics) > MAX_SUBTOPICS:
            subtopics = subtopics[:MAX_SUBTOPICS]

        # Emit progress for UI
        if progress_callback:
            progress_callback(
                f"Researching {len(subtopics)} subtopics in parallel",
                None,
                {
                    "phase": "sub_research",
                    "type": "milestone",
                    "subtopics": subtopics,
                },
            )

        current_date = datetime.now(UTC).strftime("%Y-%m-%d")
        subagent_prompt = (
            f"You are a focused research assistant. Today's date: {current_date}. "
            "Search thoroughly and return a concise factual summary. "
            "Reference sources by their [N] index numbers. "
            "Do NOT ask clarifying questions — provide your findings directly."
        )

        def run_subagent(topic: str) -> str:
            # Each subagent gets its own tool instances (thread safety)
            sub_web_search = _make_web_search_tool(
                search_engine_name,
                model,
                settings_snapshot,
                collector,
                programmatic_mode=programmatic_mode,
            )
            sub_tools = [sub_web_search]
            sub_fetch = build_fetch_tool(
                fetch_mode,
                collector,
                model=model,
                overall_query=overall_query,
                settings_snapshot=settings_snapshot,
                egress_context=egress_context,
            )
            if sub_fetch is not None:
                sub_tools.append(sub_fetch)
            try:
                # NOTE: create_agent() binds tools to the BASE LLM
                # (bind_tools resolves via ProcessingLLMWrapper.__getattr__),
                # bypassing the wrapper's <think>-tag stripping. Reasoning-model
                # output from this agent loop is NOT think-stripped (cosmetic
                # leak only; does not crash). Known limitation — see
                # ProcessingLLMWrapper in config/llm_config.py. Do NOT "fix" by
                # re-wrapping per call; the proper fix is a Runnable wrapper
                # subclass so bind_tools stays wrapped.
                agent = create_agent(
                    model=model,
                    tools=sub_tools,
                    system_prompt=subagent_prompt,
                )
                result = agent.invoke(
                    {"messages": [{"role": "user", "content": topic}]},
                    {"recursion_limit": max_sub_iterations * 2 + 1},
                )
                messages = result.get("messages", [])
                if messages:
                    last = messages[-1]
                    content = getattr(last, "content", str(last))
                    if content:
                        return content
                return f"No findings for: {topic}"
            except GraphRecursionError:
                return f"Research on '{topic}' reached iteration limit. Partial findings above."
            except Exception as exc:
                logger.exception(f"Subagent failed for: {topic[:80]}")
                return _scrub_tool_error(f"Research on '{topic}' failed: {exc}")

        # Capture the lead thread's search context (it carries the user's DB
        # password) so each pool worker can open the per-user ENCRYPTED database
        # when a subagent re-creates a search engine / registers the user's
        # document collections. stdlib ThreadPoolExecutor does NOT propagate the
        # ContextVar — without this, a collection/library primary fails inside a
        # subagent with "Unknown search engine 'collection_…'". This is the same
        # gap sibling strategies (source_based, focused_iteration) close with
        # @preserve_research_context; captured ONCE here on the lead thread.
        captured_search_context = get_search_context()

        def _run_subagent_with_egress(topic: str) -> str:
            # threading.local is NOT inherited by ThreadPoolExecutor workers,
            # so the PEP-578 audit-hook backstop armed on the parent thread is
            # inactive here. Re-arm it for the subagent's lifetime (LLM,
            # web-search and fetch connects) so the secondary egress net has
            # parity with the main thread under PRIVATE_ONLY/STRICT. The
            # primary snapshot-based PEPs already gate these calls; this
            # restores defense-in-depth. active_egress_context is a no-op when
            # egress_context is None (fail-open build).
            from ...security.egress.audit_hook import active_egress_context

            with active_egress_context(egress_context):
                # search_context sets the password ContextVar for this worker
                # and clears it on exit (preventing leak across pooled tasks).
                if captured_search_context is not None:
                    with search_context(captured_search_context):
                        return run_subagent(topic)
                return run_subagent(topic)

        ordered_results: dict[str, str] = {}
        num_workers = min(MAX_SUBAGENT_WORKERS, len(subtopics))
        with ThreadPoolExecutor(max_workers=num_workers) as executor:
            futures = {
                executor.submit(_run_subagent_with_egress, t): t
                for t in subtopics
            }
            try:
                for future in as_completed(
                    futures, timeout=SUBAGENT_TIMEOUT_SECONDS
                ):
                    topic = futures[future]
                    try:
                        ordered_results[topic] = future.result(
                            timeout=SUBAGENT_TIMEOUT_SECONDS
                        )
                    except TimeoutError:
                        logger.warning(f"Subagent timed out for: {topic[:80]}")
                        ordered_results[topic] = (
                            f"Research on '{topic}' timed out."
                        )
                    except Exception as exc:
                        logger.exception(f"Subagent failed for: {topic[:80]}")
                        ordered_results[topic] = _scrub_tool_error(
                            f"Research on '{topic}' failed: {exc}"
                        )
            except TimeoutError:
                # as_completed itself timed out — some futures didn't finish
                for future, topic in futures.items():
                    if topic not in ordered_results:
                        logger.warning(
                            f"Subagent timed out (overall): {topic[:80]}"
                        )
                        ordered_results[topic] = (
                            f"Research on '{topic}' timed out after "
                            f"{SUBAGENT_TIMEOUT_SECONDS}s."
                        )

        # Return results in original order
        parts = []
        for topic in subtopics:
            parts.append(
                f"## {topic}\n{ordered_results.get(topic, 'No results')}"
            )
        return "\n\n---\n\n".join(parts)

    return research_subtopic


# ---------------------------------------------------------------------------
# Strategy class
# ---------------------------------------------------------------------------


class LangGraphAgentStrategy(BaseSearchStrategy):
    """Research strategy using LangGraph agents with parallel subagent support.

    The lead agent autonomously decides what to search, when to dig deeper
    (via subagents), and when to synthesize — replacing the manual ReAct loop
    in the MCP strategy.
    """

    def __init__(
        self,
        model: BaseChatModel,
        search,
        citation_handler=None,
        max_iterations: int = 50,
        max_sub_iterations: int = 8,
        include_sub_research: bool = True,
        all_links_of_system: list | None = None,
        settings_snapshot: dict | None = None,
        programmatic_mode: bool = False,
        **kwargs,
    ):
        super().__init__(
            all_links_of_system=all_links_of_system,
            settings_snapshot=settings_snapshot,
            **kwargs,
        )
        self.model = model
        self.search = search
        # Whether the parent AdvancedSearchSystem is running in programmatic
        # mode (no DB metrics/rate-limit persistence). Threaded into the
        # tool factory closures so engines created per tool call inherit it.
        self.programmatic_mode = programmatic_mode
        # search.iterations (typically 1-5) controls pipeline strategies.
        # For an agent, each "iteration" is one LLM→tool round-trip, so we
        # need many more.  Treat any value below the agent minimum as "use
        # default" rather than clamping to a uselessly low number.
        self.max_iterations = (
            int(max_iterations)
            if int(max_iterations) >= MIN_ITERATIONS
            else DEFAULT_MAX_ITERATIONS
        )
        self.max_sub_iterations = int(max_sub_iterations)
        self.include_sub_research = include_sub_research
        self.citation_handler = citation_handler or CitationHandler(
            model,
            handler_type="standard",
            settings_snapshot=settings_snapshot,
        )
        self.collector = SearchResultsCollector(self.all_links_of_system)

        fetch_mode = self.get_setting(
            "search.fetch.mode", "summary_focus_query"
        )
        if fetch_mode not in FETCH_MODES:
            logger.warning(
                f"Unknown search.fetch.mode={fetch_mode!r}, falling back to "
                f"'summary_focus_query'. Valid modes: {FETCH_MODES}"
            )
            fetch_mode = "summary_focus_query"
        self.fetch_mode = fetch_mode
        logger.info(f"LangGraph agent fetch_mode={self.fetch_mode}")

        # Derive the search engine name for creating fresh instances
        self._search_engine_name = self._resolve_engine_name()

    def _resolve_engine_name(self) -> str:
        """Best-effort extraction of the configured engine name."""
        # Try settings first
        tool_setting = self.get_setting("search.tool", None)
        if tool_setting and isinstance(tool_setting, str):
            return tool_setting
        # Fall back to class name heuristic
        if self.search is not None and hasattr(self.search, "__class__"):
            name = self.search.__class__.__name__
            return name.replace("SearchEngine", "").lower()
        return "duckduckgo"

    def _get_current_engine_name(self) -> str:
        """Get the name of the currently selected search engine."""
        try:
            if hasattr(self.search, "__class__"):
                return self.search.__class__.__name__.replace(
                    "SearchEngine", ""
                ).lower()
        except Exception:
            logger.debug("Could not extract engine name from class")
        return ""

    def _display_tool_name(self, tool_name: str) -> str:
        """Return a user-friendly display name for a tool.

        ``web_search`` is a generic wrapper around the user's configured
        engine. Resolve it through the same curated ``_TOOL_DISPLAY_NAMES``
        map as the specialized search tools (keyed by ``search_<engine>``)
        so the UI shows brand-correct names like "DuckDuckGo" or
        "the web (SearXNG)" instead of the raw lowercase engine id
        (e.g. "searxng"). Other tools use the map directly.
        """
        if tool_name == "web_search":
            return _tool_display_name(f"search_{self._search_engine_name}")
        return _tool_display_name(tool_name)

    def _build_egress_context(self):
        """Construct the frozen ``EgressContext`` for this run.

        Returns ``None`` if a context can't be built (no snapshot, or
        invariant violation) — callers fall through to current behavior
        rather than crashing. Lazy import to avoid pulling the security
        module at strategy-class import time.
        """
        if not self.settings_snapshot:
            return None
        from local_deep_research.security.egress.policy import (
            PolicyDeniedError,
            context_from_snapshot,
            resolve_run_primary_engine,
        )

        try:
            # Derive the primary engine the SAME way the factory PEP does —
            # from ``search.tool`` — NOT from the engine class name. Under the
            # default ADAPTIVE scope the primary IS what resolves the concrete
            # scope, so a divergent primary here silently under-filters the
            # agent's tool list: a private collection primary classified via the
            # class heuristic ("libraryrag" -> unknown -> BOTH) left public
            # engines visible, which the factory then hard-denied mid-run
            # (scope_mismatch_private_only). resolve_run_primary_engine raises
            # ValueError when no primary is configured; this advisory filter
            # then degrades to unfiltered (the factory PEP still enforces) —
            # research_service has already failed the run closed by that point.
            primary = resolve_run_primary_engine(self.settings_snapshot)
            return context_from_snapshot(self.settings_snapshot, primary)
        except PolicyDeniedError:
            # Corrupted/invalid policy.egress_scope — re-raise so the
            # caller fails closed instead of silently running unfiltered.
            raise
        except (ValueError, KeyError, TypeError):
            logger.debug(
                "Could not build EgressContext for langgraph agent — "
                "falling back to unfiltered tool list"
            )
            return None

    def _build_tools(self, overall_query: str = "") -> list:
        """Build the LangChain tool list for the lead agent.

        ``overall_query`` is the original user query; it's threaded into
        summary-mode fetch tools so the per-page extractor sees both the
        agent's per-fetch focus and the original research question.
        """
        tools = []

        # Compute the policy context ONCE for this run. Threaded through
        # every tool builder so subagent threads — which don't inherit
        # thread-local state — get the same context as the lead agent.
        policy_ctx = self._build_egress_context()

        # Web search (always present if we have a search engine)
        if self.search is not None:
            tools.append(
                _make_web_search_tool(
                    self._search_engine_name,
                    self.model,
                    self.settings_snapshot,
                    self.collector,
                    programmatic_mode=self.programmatic_mode,
                )
            )

        # Content fetcher (returns None when fetch_mode == 'disabled')
        fetch = build_fetch_tool(
            self.fetch_mode,
            self.collector,
            model=self.model,
            overall_query=overall_query,
            settings_snapshot=self.settings_snapshot,
            egress_context=policy_ctx,
        )
        if fetch is not None:
            tools.append(fetch)

        # Specialized search engines (pre-filtered by egress policy).
        #
        # This is the core fix for the original LangGraph silent-expansion
        # complaint. The factory PEP catches engines at instantiation time,
        # but that's a runtime check — the LLM still SEES the forbidden
        # tool names in the schema and the latency of a denied tool call
        # leaks policy state. Filtering the tool list HERE means the
        # forbidden tools never reach create_agent(), and the LLM never
        # learns they exist.
        try:
            from local_deep_research.web_search_engines.search_engines_config import (
                get_available_engines,
            )
            from local_deep_research.security.egress.policy import (
                EgressScope,
                evaluate_engine,
                evaluate_retriever,
            )
            from local_deep_research.web_search_engines.retriever_registry import (
                retriever_registry,
            )

            available = get_available_engines(
                settings_snapshot=self.settings_snapshot,
            )
            current = self._get_current_engine_name()
            for name, config in available.items():
                if name == current:
                    continue

                # Per-collection usability switch (independent of egress): a
                # collection the user marked "not for the research agent" is
                # skipped here so it never appears in the agent's tool list.
                # Non-collection engines and NULL/missing flags default to
                # available, so existing behaviour is unchanged.
                if not config.get("agent_enabled", True):
                    logger.debug(
                        "specialized tool skipped: collection disabled for "
                        "the research agent",
                        engine=name,
                    )
                    continue

                # Under STRICT, register no specialized engines at all —
                # the agent gets only the primary web_search tool.
                if (
                    policy_ctx is not None
                    and policy_ctx.scope == EgressScope.STRICT
                ):
                    continue

                # Under PUBLIC_ONLY / PRIVATE_ONLY, ask the PDP whether
                # this engine fits the scope. Retrievers route to
                # evaluate_retriever (engine-PDP returns engine_unknown
                # for them); plain engines route to evaluate_engine.
                if policy_ctx is not None:
                    if config.get("is_retriever"):
                        try:
                            meta = retriever_registry.get_metadata(name)
                        except AttributeError:
                            meta = None
                        decision = evaluate_retriever(
                            name, policy_ctx, metadata=meta
                        )
                    else:
                        # Pass the engine config as metadata so a
                        # per-collection is_public classification is honored
                        # without a redundant DB lookup per collection.
                        decision = evaluate_engine(
                            name,
                            policy_ctx,
                            settings_snapshot=self.settings_snapshot,
                            metadata=config,
                        )
                    if not decision.allowed:
                        logger.bind(policy_audit=True).info(
                            "specialized tool filtered by egress policy",
                            engine=name,
                            scope=policy_ctx.scope.value,
                            reason=decision.reason,
                        )
                        continue

                desc = config.get("description", f"Search using {name}")
                strengths = config.get("strengths", [])
                if strengths:
                    desc += f" Best for: {', '.join(strengths[:2])}."
                tools.append(
                    _make_specialized_search_tool(
                        name,
                        desc,
                        self.model,
                        self.settings_snapshot,
                        self.collector,
                        programmatic_mode=self.programmatic_mode,
                    )
                )
        except Exception:
            logger.warning(
                "Failed to load specialized search engines for agent tools"
            )

        # Subagent research tool
        if self.include_sub_research:
            tools.append(
                _make_research_subtopic_tool(
                    self._search_engine_name,
                    self.model,
                    self.settings_snapshot,
                    self.collector,
                    self.max_sub_iterations,
                    progress_callback=self.progress_callback,
                    programmatic_mode=self.programmatic_mode,
                    fetch_mode=self.fetch_mode,
                    overall_query=overall_query,
                    egress_context=policy_ctx,
                )
            )

        return tools

    # -- Main entry point ---------------------------------------------------

    def analyze_topic(self, query: str) -> Dict[str, Any]:
        from langchain.agents import create_agent

        logger.info(f"LangGraph agent research: {query[:100]}")

        # Reset collector for fresh subsection call (detailed report mode)
        self.collector.reset()
        nr_of_links = len(self.all_links_of_system)

        self._update_progress(
            f'Starting agent research: "{query[:80]}"',
            5,
            {"phase": "init", "type": "milestone", "query": query[:100]},
        )
        self.check_termination()

        # Build tools (overall_query feeds summary-mode fetch tools)
        tools = self._build_tools(overall_query=query)
        if not tools:
            return self._error_result("No tools available")
        # Stash tool names for the per-step heartbeat — gives the user
        # concrete info ("from the web (SearXNG), PubMed, …") instead of
        # a vague spinner while the LLM picks its next move. The raw ids
        # are mapped to friendly names at render time via
        # ``_display_tool_name``.
        self._tool_names = [getattr(t, "name", "?") for t in tools]

        # Build system prompt — fetch_line wording mirrors the active mode
        # so the agent isn't told to use a tool that doesn't exist.
        current_date = datetime.now(UTC).strftime("%Y-%m-%d")
        if self.fetch_mode == "disabled":
            fetch_line = (
                "3. Rely on search snippets — full-page fetching is disabled "
                "for this run.\n"
            )
        elif self.fetch_mode in ("summary_focus", "summary_focus_query"):
            fetch_line = (
                "3. Use fetch_content(url, focus) when snippets aren't enough; "
                "always pass the specific question or claim you want answered "
                "as ``focus`` so the tool returns only the relevant facts.\n"
            )
        else:  # full
            fetch_line = "3. Use fetch_content to read full pages when snippets aren't enough.\n"
        # Build the policy addendum once so the system prompt can carry
        # explicit guidance to the LLM about which tools actually exist.
        # Closing the timing-leak attack requires both halves: the tool
        # list is pre-filtered above, AND the LLM is told what's
        # available, so it doesn't waste tokens probing for forbidden
        # engines and the latency of denial paths doesn't leak policy.
        policy_addendum = ""
        try:
            from local_deep_research.security.egress.policy import (
                EgressScope,
                PolicyDeniedError,
            )

            ctx = self._build_egress_context()
            if ctx is not None and ctx.scope == EgressScope.STRICT:
                policy_addendum = (
                    "\nRESTRICTED MODE: only the primary search tool is "
                    f"available ({ctx.primary_engine}). Do NOT reference or "
                    "attempt other search_* tools — they do not exist in "
                    "this session and will not work. Use web_search and "
                    "research_subtopic for everything.\n"
                )
            elif ctx is not None and ctx.scope == EgressScope.PRIVATE_ONLY:
                policy_addendum = (
                    "\nPRIVATE-ONLY MODE: public search engines (arxiv, "
                    "pubmed, brave, etc.) are not available in this "
                    "session. Use only local search tools.\n"
                )
            elif ctx is not None and ctx.scope == EgressScope.PUBLIC_ONLY:
                policy_addendum = (
                    "\nPUBLIC-ONLY MODE: local search tools (library, "
                    "collection, paperless) are not available in this "
                    "session.\n"
                )
        except PolicyDeniedError:
            # Corrupt/unknown scope must fail closed, never run unfiltered.
            # In practice _build_tools() (called above at the top of
            # analyze_topic) already raised this for the same snapshot, so
            # we never reach here with a bad scope — but re-raise rather
            # than swallow, so this stays correct if the call order changes.
            raise
        except Exception:
            logger.debug(
                "Could not derive policy addendum for system prompt — "
                "agent will see the unmodified prompt"
            )

        system_prompt = (
            f"You are a research assistant writing a research report. Today's date: {current_date}.\n"
            "This is NOT a chat conversation. Your only job is to research the "
            "given topic and produce a comprehensive, well-cited report.\n"
            "Do NOT ask clarifying questions, do NOT ask the user anything, "
            "do NOT offer to help further — just research and report.\n"
            "You MUST search the web before answering — never answer from memory alone.\n\n"
            "Strategy:\n"
            "1. Start with web_search for initial exploration.\n"
            "2. For complex multi-faceted questions, use research_subtopic to "
            "investigate specific aspects in parallel (pass 2-5 focused questions).\n"
            f"{fetch_line}"
            "4. Use search_[engine] tools for domain-specific searches "
            "(search_arxiv for science, search_pubmed for medical, etc.).\n"
            "5. When you have enough information, provide a comprehensive answer "
            "citing sources as [1], [2], etc.\n"
            f"{policy_addendum}"
        )

        # Create agent — may fail if model doesn't support tool calling.
        # NOTE: create_agent() binds tools to the BASE LLM (bind_tools resolves
        # via ProcessingLLMWrapper.__getattr__), bypassing the wrapper's
        # <think>-tag stripping. Reasoning-model output from this agent loop is
        # NOT think-stripped (cosmetic leak only; does not crash). Known
        # limitation — see ProcessingLLMWrapper in config/llm_config.py.
        try:
            agent = create_agent(
                model=self.model,
                tools=tools,
                system_prompt=system_prompt,
            )
        except Exception as exc:
            logger.exception("Failed to create LangGraph agent")
            return self._error_result(
                _scrub_tool_error(
                    f"Failed to create agent (model may not "
                    f"support tool calling): {exc}"
                )
            )

        # Stream agent execution
        effective_max = max(MIN_ITERATIONS, self.max_iterations)
        config = {"recursion_limit": effective_max * 2 + 1}
        iteration = 0
        final_content = ""
        agent_messages: list = []

        try:
            for chunk in agent.stream(
                {"messages": [{"role": "user", "content": query}]},
                config,
                stream_mode="updates",
            ):
                self.check_termination()

                if "agent" in chunk or "model" in chunk:
                    node_key = "agent" if "agent" in chunk else "model"
                    iteration += 1
                    progress = 10 + int((iteration / effective_max) * 75)
                    msgs = chunk[node_key].get("messages", [])
                    for msg in msgs:
                        if isinstance(msg, AIMessage):
                            agent_messages.append(msg)
                            content = msg.content or ""
                            tool_calls = getattr(msg, "tool_calls", [])

                            # Surface the model's *thinking* output (the
                            # <think>…</think> reasoning) when reasoning
                            # mode is on. langchain-ollama puts the
                            # discarded thinking content into
                            # additional_kwargs["reasoning_content"]; we
                            # emit it as agent_reasoning so the thinking
                            # bubble shows the agent's actual rationale
                            # ("I should search for X because…") right
                            # before the next tool call fires. This is
                            # per-step (one emit per LLM round) —
                            # token-level streaming would require switching
                            # langgraph to stream_mode=["updates",
                            # "messages"] and capturing chunks inside agent
                            # nodes, which is a larger change.
                            reasoning_text = ""
                            if getattr(msg, "additional_kwargs", None):
                                reasoning_text = str(
                                    msg.additional_kwargs.get(
                                        "reasoning_content", ""
                                    )
                                    or ""
                                ).strip()
                            # Fall back to msg.content when the model
                            # emitted prose alongside tool_calls (rare for
                            # tool-calling LLMs — most emit only the tool
                            # call), but harmless when both apply.
                            if not reasoning_text and content and tool_calls:
                                reasoning_text = str(content).strip()
                            if reasoning_text:
                                self._update_progress(
                                    reasoning_text[:280],
                                    min(85, progress),
                                    {
                                        "phase": "agent_reasoning",
                                        "iteration": iteration,
                                    },
                                )

                            if tool_calls:
                                for tc in tool_calls:
                                    tc_args = tc.get("args", {})
                                    raw_name = tc["name"]
                                    display_name = self._display_tool_name(
                                        raw_name
                                    )
                                    # `fetch_url` carries a URL arg; the
                                    # search tools carry a query arg.
                                    # Either way, show the meaningful arg
                                    # in quotes so the user sees what the
                                    # agent is actually looking up.
                                    if raw_name == "fetch_url":
                                        target = str(tc_args.get("url", ""))[
                                            :80
                                        ]
                                        msg_text = f'📖 Reading {display_name}: "{target}"'
                                    elif raw_name == "research_subtopic":
                                        # Tool signature is `subtopics: list[str]`.
                                        # Accept either key for forward-compat
                                        # and stringify list as a comma list.
                                        raw_sub = tc_args.get(
                                            "subtopics",
                                            tc_args.get(
                                                "subtopic",
                                                tc_args.get("query", ""),
                                            ),
                                        )
                                        if isinstance(raw_sub, list):
                                            sub = ", ".join(
                                                str(s) for s in raw_sub
                                            )[:80]
                                        else:
                                            sub = str(raw_sub)[:80]
                                        msg_text = f'🔬 Investigating subtopic: "{sub}"'
                                    else:
                                        # Use a loop-local name here — do NOT
                                        # reassign the `query` parameter, which
                                        # is still needed downstream by
                                        # _synthesize_from_collector()/_finalize()
                                        # as the original research question.
                                        tc_query = str(
                                            tc_args.get(
                                                "query",
                                                tc_args.get("url", ""),
                                            )
                                        )[:80]
                                        msg_text = f'🔍 Searching {display_name}: "{tc_query}"'
                                    self._update_progress(
                                        msg_text,
                                        min(85, progress),
                                        {
                                            "phase": "tool_call",
                                            # Keep the stable tool id in
                                            # metadata; the friendly label
                                            # already lives in msg_text.
                                            "tool": raw_name,
                                            "iteration": iteration,
                                        },
                                    )
                            elif content:
                                # No tool calls = final answer
                                final_content = content

                elif "tools" in chunk:
                    msgs = chunk["tools"].get("messages", [])
                    for msg in msgs:
                        tool_name = getattr(msg, "name", "tool")
                        display_name = self._display_tool_name(tool_name)
                        preview = str(getattr(msg, "content", ""))[
                            :150
                        ].replace("\n", " ")
                        self._update_progress(
                            f"📄 From {display_name}: {preview}",
                            min(
                                85,
                                10 + int((iteration / effective_max) * 75) + 3,
                            ),
                            # Keep the stable tool id in metadata; the
                            # friendly label already lives in the message.
                            {"phase": "observation", "tool": tool_name},
                        )
                    # After every tool result, the agent immediately re-
                    # invokes the model to decide the next step. For
                    # thinking-mode LLMs (Qwen 3.x, deepseek-r1, etc.)
                    # that step can take 30+ seconds of silent <think>
                    # generation that gets stripped before display —
                    # leaving the last displayed line stale ("Result from
                    # web_search …") with no indication the agent is still
                    # working.
                    # Emit a contextual heartbeat so the user gets a real
                    # sense of progress (which iteration, how many sources
                    # collected, which tools are available) instead of
                    # a generic "Choosing next step…" spinner.
                    sources_so_far = len(self.all_links_of_system)
                    tool_count = len(getattr(self, "_tool_names", []) or [])
                    if sources_so_far == 0:
                        heartbeat = (
                            f"Step {iteration} · planning approach "
                            f"with {tool_count} research tool"
                            f"{'s' if tool_count != 1 else ''} available…"
                        )
                    else:
                        # Show up to 3 representative tool names so the
                        # user sees what the agent might pick next without
                        # the line ballooning when many specialised
                        # engines are enabled.
                        names = getattr(self, "_tool_names", []) or []
                        sample = ", ".join(
                            self._display_tool_name(n) for n in names[:3]
                        )
                        more = (
                            f" +{len(names) - 3} more" if len(names) > 3 else ""
                        )
                        heartbeat = (
                            f"Step {iteration} · {sources_so_far} source"
                            f"{'s' if sources_so_far != 1 else ''} gathered · "
                            f"selecting next action from {sample}{more}…"
                        )
                    self._update_progress(
                        heartbeat,
                        min(
                            85,
                            10 + int((iteration / effective_max) * 75) + 4,
                        ),
                        {"phase": "agent_thinking", "iteration": iteration},
                    )

        except GraphRecursionError:
            logger.warning(
                "LangGraph agent hit recursion limit, synthesizing partial results"
            )
            if not final_content:
                final_content = self._synthesize_from_collector(query)
        except Exception as exc:
            logger.exception("LangGraph agent error")
            if not final_content:
                if self.collector.results:
                    final_content = self._synthesize_from_collector(query)
                else:
                    return self._error_result(self._format_agent_error(exc))

        if not final_content:
            if self.collector.results:
                final_content = self._synthesize_from_collector(query)
            else:
                final_content = (
                    "Research could not produce results. Try a different query."
                )

        return self._finalize(
            query, final_content, iteration, nr_of_links, agent_messages
        )

    # -- Helpers ------------------------------------------------------------

    def _synthesize_from_collector(self, query: str) -> str:
        """Fallback synthesis when the agent was cut short."""
        results = self.collector.results
        if not results:
            return "Research could not be completed within the iteration limit."
        summaries = []
        for r in results[:20]:
            summaries.append(
                f"[{r.get('index', '?')}] {r.get('title', '')}: "
                f"{r.get('snippet', '')}"
            )
        prompt = (
            f"Synthesize a comprehensive answer to: {query}\n\n"
            f"Based on these sources:\n" + "\n".join(summaries)
        )
        try:
            response = self.model.invoke(prompt)
            return (
                response.content
                if hasattr(response, "content")
                else str(response)
            )
        except Exception as exc:
            logger.exception("Fallback synthesis failed")
            return _scrub_tool_error(
                f"Research collected {len(results)} sources but "
                f"synthesis failed: {exc}"
            )

    def _finalize(
        self,
        query: str,
        final_answer: str,
        iteration: int,
        nr_of_links: int,
        agent_messages: list,
    ) -> Dict[str, Any]:
        """Apply citation handling and build the return dict."""
        self._update_progress(
            f"Synthesizing {len(self.collector.results)} sources with citations",
            90,
            {"phase": "synthesis", "type": "milestone"},
        )

        all_search_results = self.collector.results
        synthesized_content = final_answer
        documents: list = []

        # Citation handling — only if we have results
        if all_search_results:
            try:
                citation_result = self.citation_handler.analyze_followup(
                    query,
                    all_search_results,
                    previous_knowledge=final_answer,
                    nr_of_links=nr_of_links,
                )
                if isinstance(citation_result, dict):
                    synthesized_content = citation_result.get(
                        "content", citation_result.get("response", final_answer)
                    )
                    documents = citation_result.get("documents", [])
            except Exception:
                logger.warning(
                    "Citation handler failed, using raw agent answer"
                )

        # Format sources — delegate to base helper
        formatted_output = self._format_citations(
            synthesized_content, all_search_results
        )

        # Build reasoning trace from agent messages
        reasoning_trace = []
        for msg in agent_messages:
            entry: Dict[str, Any] = {"role": "assistant"}
            if hasattr(msg, "content") and msg.content:
                entry["content"] = msg.content
            tool_calls = getattr(msg, "tool_calls", [])
            if tool_calls:
                entry["tool_calls"] = [
                    {"name": tc.get("name"), "args": tc.get("args", {})}
                    for tc in tool_calls
                ]
            reasoning_trace.append(entry)

        self._update_progress(
            "Research complete",
            100,
            {"phase": "complete", "type": "milestone", "iterations": iteration},
        )

        return {
            "findings": [
                {
                    "content": synthesized_content,
                    "question": query,
                    "search_results": all_search_results,
                    "documents": documents,
                }
            ],
            "iterations": iteration,
            "questions": {},
            "formatted_findings": formatted_output,
            "current_knowledge": synthesized_content,
            "sources": list(set(self.collector.sources)),
            "search_results": all_search_results,
            "documents": documents,
            "reasoning_trace": reasoning_trace,
            "error": None,
        }

    @staticmethod
    def _format_agent_error(exc: BaseException) -> str:
        """Prefix the exception type so downstream rendering (and the
        `ErrorReportGenerator` pattern map) have a consistent shape to match
        on. The bare `str(exc)` produced by the catch-all loses the type,
        which makes deep LangChain / LangGraph failures hard to recognise.
        """
        # Scrub credentials before this error is rendered to the user. The
        # "Agent error: <Type>:" prefix stays at the front (no secrets, ahead
        # of any truncation) so the ErrorReportGenerator pattern map still
        # matches on the exception type.
        return _scrub_tool_error(f"Agent error: {type(exc).__name__}: {exc}")

    def _error_result(self, error: str) -> Dict[str, Any]:
        logger.error(f"LangGraph agent strategy error: {error}")
        self._update_progress(
            f"Error: {error}",
            100,
            {"phase": "error", "error": error, "status": "failed"},
        )
        return {
            "findings": [],
            "iterations": 0,
            "questions": {},
            "formatted_findings": f"Error: {error}",
            "current_knowledge": "",
            "sources": [],
            "search_results": [],
            "documents": [],
            "reasoning_trace": [],
            "error": error,
        }

    def close(self):
        """No persistent resources to clean up."""
        pass
