"""
Tests for the LangGraph agent research strategy.

Tests cover:
- SearchResultsCollector thread safety and behavior
- Tool factory functions
- Strategy instantiation and configuration
- Citation offset handling for detailed report mode
- Error handling paths
- Egress-scope tool filtering (TestEgressScopeFiltering at end of file)
"""

import threading
from unittest.mock import MagicMock, patch

import pytest


# ---------------------------------------------------------------------------
# SearchResultsCollector tests
# ---------------------------------------------------------------------------


class TestSearchResultsCollector:
    """Tests for the thread-safe SearchResultsCollector."""

    def _make_collector(self, all_links=None):
        from local_deep_research.advanced_search_system.strategies.langgraph_agent_strategy import (
            SearchResultsCollector,
        )

        links = all_links if all_links is not None else []
        return SearchResultsCollector(links), links

    def test_add_results_indexes_correctly(self):
        collector, all_links = self._make_collector()
        results = [
            {"title": "A", "link": "http://a.com", "snippet": "a"},
            {"title": "B", "link": "http://b.com", "snippet": "b"},
        ]
        start = collector.add_results(results, engine_name="test")

        assert start == 0
        assert len(collector.results) == 2
        assert collector.results[0]["index"] == "1"
        assert collector.results[1]["index"] == "2"

    def test_add_results_continues_indexing(self):
        collector, _ = self._make_collector()
        collector.add_results(
            [{"title": "A", "link": "http://a.com", "snippet": "a"}],
            engine_name="test",
        )
        start = collector.add_results(
            [{"title": "B", "link": "http://b.com", "snippet": "b"}],
            engine_name="test",
        )

        assert start == 1
        assert collector.results[1]["index"] == "2"

    def test_add_results_normalizes_url_to_link(self):
        collector, _ = self._make_collector()
        results = [{"title": "A", "url": "http://a.com", "snippet": "a"}]
        collector.add_results(results)

        assert "link" in collector.results[0]
        assert collector.results[0]["link"] == "http://a.com"

    def test_add_results_preserves_existing_link(self):
        collector, _ = self._make_collector()
        results = [
            {
                "title": "A",
                "link": "http://link.com",
                "url": "http://url.com",
                "snippet": "a",
            }
        ]
        collector.add_results(results)

        assert collector.results[0]["link"] == "http://link.com"

    def test_add_results_sets_source_engine(self):
        collector, _ = self._make_collector()
        results = [{"title": "A", "link": "http://a.com", "snippet": "a"}]
        collector.add_results(results, engine_name="arxiv")

        assert collector.results[0]["source_engine"] == "arxiv"

    def test_add_results_appends_to_all_links(self):
        all_links = []
        collector, _ = self._make_collector(all_links)
        results = [{"title": "A", "link": "http://a.com", "snippet": "a"}]
        collector.add_results(results)

        assert len(all_links) == 1
        assert all_links[0]["index"] == "1"

    def test_reset_clears_results_but_not_all_links(self):
        all_links = []
        collector, _ = self._make_collector(all_links)
        collector.add_results(
            [{"title": "A", "link": "http://a.com", "snippet": "a"}]
        )
        assert len(collector.results) == 1
        assert len(all_links) == 1

        collector.reset()

        assert len(collector.results) == 0
        assert len(collector.sources) == 0
        # all_links must NOT be cleared
        assert len(all_links) == 1

    def test_sources_tracks_links(self):
        collector, _ = self._make_collector()
        collector.add_results(
            [
                {"title": "A", "link": "http://a.com", "snippet": "a"},
                {"title": "B", "link": "http://b.com", "snippet": "b"},
            ]
        )

        assert set(collector.sources) == {"http://a.com", "http://b.com"}

    def test_add_results_does_not_mutate_input(self):
        collector, _ = self._make_collector()
        original = {"title": "A", "link": "http://a.com", "snippet": "a"}
        collector.add_results([original])

        # Original dict should NOT have index/source_engine added
        assert "index" not in original

    def test_empty_results_returns_current_length(self):
        collector, _ = self._make_collector()
        collector.add_results(
            [{"title": "A", "link": "http://a.com", "snippet": "a"}]
        )
        start = collector.add_results([])
        assert start == 1

    def test_thread_safety_no_duplicate_indices(self):
        """Multiple threads adding results should never produce duplicate indices."""
        collector, _ = self._make_collector()
        results_per_thread = [
            {"title": f"T{i}", "link": f"http://{i}.com", "snippet": f"s{i}"}
            for i in range(5)
        ]
        errors = []

        def add_batch(thread_id):
            try:
                collector.add_results(
                    [dict(r) for r in results_per_thread],
                    engine_name=f"thread-{thread_id}",
                )
            except Exception as exc:
                errors.append(exc)

        threads = [
            threading.Thread(target=add_batch, args=(i,)) for i in range(4)
        ]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        assert not errors
        all_results = collector.results
        assert len(all_results) == 20  # 4 threads × 5 results
        indices = [r["index"] for r in all_results]
        assert len(indices) == len(set(indices)), "Duplicate indices found!"


# ---------------------------------------------------------------------------
# Format results helper
# ---------------------------------------------------------------------------


class TestFormatResults:
    def test_format_results_basic(self):
        from local_deep_research.advanced_search_system.strategies.langgraph_agent_strategy import (
            _format_results,
        )

        results = [
            {
                "title": "Test",
                "link": "http://test.com",
                "snippet": "A snippet",
            },
        ]
        output = _format_results(results, start_idx=0)
        assert "[1]" in output
        assert "Test" in output
        assert "http://test.com" in output
        assert "A snippet" in output

    def test_format_results_offset(self):
        from local_deep_research.advanced_search_system.strategies.langgraph_agent_strategy import (
            _format_results,
        )

        results = [
            {"title": "Test", "link": "http://test.com", "snippet": "snip"},
        ]
        output = _format_results(results, start_idx=5)
        assert "[6]" in output

    def test_format_empty_returns_no_results(self):
        from local_deep_research.advanced_search_system.strategies.langgraph_agent_strategy import (
            _format_results,
        )

        assert _format_results([], 0) == "No results."


# ---------------------------------------------------------------------------
# Strategy instantiation and configuration
# ---------------------------------------------------------------------------


class TestLangGraphAgentStrategy:
    """Test strategy construction and configuration."""

    def _make_strategy(self, **overrides):
        from local_deep_research.advanced_search_system.strategies.langgraph_agent_strategy import (
            LangGraphAgentStrategy,
        )

        defaults = {
            "model": MagicMock(),
            "search": MagicMock(),
            "all_links_of_system": [],
            "settings_snapshot": {"search.tool": {"value": "duckduckgo"}},
        }
        defaults.update(overrides)
        return LangGraphAgentStrategy(**defaults)

    def test_basic_instantiation(self):
        strategy = self._make_strategy()
        assert strategy is not None
        assert hasattr(strategy, "analyze_topic")
        assert hasattr(strategy, "collector")

    def test_format_agent_error_scrubs_credentials(self):
        """_format_agent_error is rendered to the user, so it must scrub
        credentials from the exception text — while keeping the
        'Agent error: <Type>:' prefix the ErrorReportGenerator pattern map
        matches on (credential-leak follow-up to #4625)."""
        from local_deep_research.advanced_search_system.strategies.langgraph_agent_strategy import (
            LangGraphAgentStrategy,
        )

        exc = RuntimeError(
            "LLM call failed: https://api.example.com/v1?api_key=SECRETKEY123"
        )
        out = LangGraphAgentStrategy._format_agent_error(exc)

        assert "SECRETKEY123" not in out  # credential scrubbed
        assert out.startswith("Agent error: RuntimeError:")  # type prefix kept

    def test_format_agent_error_keeps_categorizable_token_past_200_chars(self):
        """The larger (500) cap for tool/agent errors keeps the categorizable
        signal that can sit deep in a long exception message — the 200-char
        HTTP-client default would truncate it and degrade ErrorReporter
        categorization to 'unknown' (#4633)."""
        from local_deep_research.advanced_search_system.strategies.langgraph_agent_strategy import (
            LangGraphAgentStrategy,
        )

        exc = RuntimeError(("x" * 230) + " Connection refused")
        out = LangGraphAgentStrategy._format_agent_error(exc)

        # The token sits past char 200; it survives the 500 cap (a 200 cap
        # would drop it).
        assert "Connection refused" in out

    def test_default_params(self):
        strategy = self._make_strategy()
        assert strategy.max_iterations == 50
        assert strategy.max_sub_iterations == 8
        assert strategy.include_sub_research is True

    def test_custom_params(self):
        strategy = self._make_strategy(
            max_iterations=50, max_sub_iterations=3, include_sub_research=False
        )
        assert strategy.max_iterations == 50
        assert strategy.max_sub_iterations == 3
        assert strategy.include_sub_research is False

    def test_low_max_iterations_uses_default(self):
        """Pipeline-style low values (e.g. search.iterations=3) should not
        constrain the agent — it needs many more ReAct cycles."""
        strategy = self._make_strategy(max_iterations=3)
        assert strategy.max_iterations == 50  # DEFAULT_MAX_ITERATIONS

    def test_super_init_called_with_kwargs(self):
        """Verify base class attributes are set correctly."""
        all_links = [{"existing": True}]
        strategy = self._make_strategy(all_links_of_system=all_links)
        assert strategy.all_links_of_system is all_links

    def test_collector_shares_all_links_reference(self):
        all_links = []
        strategy = self._make_strategy(all_links_of_system=all_links)
        strategy.collector.add_results(
            [{"title": "T", "link": "http://t.com", "snippet": "s"}]
        )
        assert len(all_links) == 1

    def test_engine_name_from_settings(self):
        strategy = self._make_strategy(
            settings_snapshot={"search.tool": {"value": "brave"}}
        )
        assert strategy._search_engine_name == "brave"

    def test_engine_name_from_settings_string(self):
        strategy = self._make_strategy(
            settings_snapshot={"search.tool": "searxng"}
        )
        assert strategy._search_engine_name == "searxng"

    def test_engine_name_fallback_to_class(self):
        mock_search = MagicMock()
        mock_search.__class__.__name__ = "DuckDuckGoSearchEngine"
        strategy = self._make_strategy(search=mock_search, settings_snapshot={})
        assert strategy._search_engine_name == "duckduckgo"

    def test_display_tool_name_web_search_uses_curated_engine_name(self):
        """``web_search`` renders the configured engine through the curated
        display-name map, with brand-correct casing rather than the raw
        lowercase id."""
        strategy = self._make_strategy(
            settings_snapshot={"search.tool": {"value": "duckduckgo"}}
        )
        assert strategy._display_tool_name("web_search") == "DuckDuckGo"

    def test_display_tool_name_web_search_searxng(self):
        strategy = self._make_strategy(
            settings_snapshot={"search.tool": {"value": "searxng"}}
        )
        assert strategy._display_tool_name("web_search") == "the web (SearXNG)"

    def test_display_tool_name_web_search_multiword_engine(self):
        """Multi-word engine ids resolve to their curated display name."""
        strategy = self._make_strategy(
            settings_snapshot={"search.tool": {"value": "semantic_scholar"}}
        )
        assert strategy._display_tool_name("web_search") == "Semantic Scholar"

    def test_display_tool_name_web_search_unknown_engine_titlecased(self):
        """Engines absent from the curated map fall back to a cleaned,
        title-cased name — never the raw lowercase id."""
        strategy = self._make_strategy(
            settings_snapshot={"search.tool": {"value": "tavily"}}
        )
        assert strategy._display_tool_name("web_search") == "Tavily"

    def test_display_tool_name_specialized_tool_uses_map(self):
        """Non-web_search tools keep their curated display name."""
        strategy = self._make_strategy()
        assert strategy._display_tool_name("search_pubmed") == "PubMed"


# ---------------------------------------------------------------------------
# Citation offset for detailed report mode
# ---------------------------------------------------------------------------


class TestCitationOffset:
    """Test that nr_of_links is handled correctly across multiple calls."""

    def _make_strategy(self):
        from local_deep_research.advanced_search_system.strategies.langgraph_agent_strategy import (
            LangGraphAgentStrategy,
        )

        model = MagicMock()
        model.invoke = MagicMock(
            return_value=MagicMock(content="Synthesized answer")
        )
        return LangGraphAgentStrategy(
            model=model,
            search=MagicMock(),
            all_links_of_system=[],
            settings_snapshot={"search.tool": {"value": "mock"}},
        )

    def test_collector_reset_on_analyze_topic(self):
        """Collector should be reset at the start of each analyze_topic call."""
        strategy = self._make_strategy()

        # Pre-populate collector
        strategy.collector.add_results(
            [{"title": "Old", "link": "http://old.com", "snippet": "old"}]
        )
        assert len(strategy.collector.results) == 1

        # analyze_topic should reset the collector
        with patch(
            "local_deep_research.advanced_search_system.strategies.langgraph_agent_strategy.LangGraphAgentStrategy._build_tools",
            return_value=[],
        ):
            result = strategy.analyze_topic("test query")

        # Collector should have been reset (even though _build_tools returned empty)
        # reset() happens before _build_tools, so the error path still resets
        assert result["error"] is not None  # error because no tools
        assert len(strategy.collector.results) == 0  # verify reset happened

    def test_all_links_accumulates_across_calls(self):
        """all_links_of_system should grow across calls, not reset."""
        strategy = self._make_strategy()
        all_links = strategy.all_links_of_system

        strategy.collector.add_results(
            [{"title": "A", "link": "http://a.com", "snippet": "a"}]
        )
        assert len(all_links) == 1

        strategy.collector.reset()

        strategy.collector.add_results(
            [{"title": "B", "link": "http://b.com", "snippet": "b"}]
        )
        assert len(all_links) == 2

    def test_citation_indices_unique_across_sections(self):
        """After reset, new results should get globally unique indices
        (not restart from 1) so detailed report citations don't collide."""
        strategy = self._make_strategy()

        # Section 1: adds 2 results → indices "1", "2"
        strategy.collector.add_results(
            [
                {"title": "A", "link": "http://a.com", "snippet": "a"},
                {"title": "B", "link": "http://b.com", "snippet": "b"},
            ]
        )
        assert strategy.all_links_of_system[0]["index"] == "1"
        assert strategy.all_links_of_system[1]["index"] == "2"

        # Simulate new section: reset per-call state
        strategy.collector.reset()

        # Section 2: should continue from "3", not restart at "1"
        strategy.collector.add_results(
            [
                {"title": "C", "link": "http://c.com", "snippet": "c"},
                {"title": "D", "link": "http://d.com", "snippet": "d"},
            ]
        )
        assert strategy.all_links_of_system[2]["index"] == "3"
        assert strategy.all_links_of_system[3]["index"] == "4"


# ---------------------------------------------------------------------------
# Error handling
# ---------------------------------------------------------------------------


class TestErrorHandling:
    """Test error paths return proper error dicts."""

    def _make_strategy(self):
        from local_deep_research.advanced_search_system.strategies.langgraph_agent_strategy import (
            LangGraphAgentStrategy,
        )

        return LangGraphAgentStrategy(
            model=MagicMock(),
            search=MagicMock(),
            all_links_of_system=[],
            settings_snapshot={"search.tool": {"value": "mock"}},
        )

    def test_error_result_structure(self):
        strategy = self._make_strategy()
        result = strategy._error_result("something broke")

        assert result["error"] == "something broke"
        assert result["findings"] == []
        assert result["iterations"] == 0
        assert result["current_knowledge"] == ""
        assert isinstance(result["reasoning_trace"], list)

    def test_no_tools_returns_error(self):
        strategy = self._make_strategy()
        with patch(
            "local_deep_research.advanced_search_system.strategies.langgraph_agent_strategy.LangGraphAgentStrategy._build_tools",
            return_value=[],
        ):
            result = strategy.analyze_topic("test")

        assert result["error"] is not None
        assert "No tools" in result["error"]

    def test_agent_creation_failure_returns_error(self):
        strategy = self._make_strategy()
        with (
            patch(
                "local_deep_research.advanced_search_system.strategies.langgraph_agent_strategy.LangGraphAgentStrategy._build_tools",
                return_value=[MagicMock()],
            ),
            patch(
                "langchain.agents.create_agent",
                side_effect=ValueError("Model doesn't support tools"),
            ),
        ):
            result = strategy.analyze_topic("test")

        assert result["error"] is not None
        assert "tool calling" in result["error"]

    def test_format_agent_error_includes_exception_type(self):
        from local_deep_research.advanced_search_system.strategies.langgraph_agent_strategy import (
            LangGraphAgentStrategy,
        )

        msg = LangGraphAgentStrategy._format_agent_error(ValueError("boom"))

        assert "ValueError" in msg
        assert "boom" in msg


# ---------------------------------------------------------------------------
# Factory integration
# ---------------------------------------------------------------------------


class TestFactoryIntegration:
    """Test that the strategy integrates with the factory correctly."""

    def test_factory_creates_langgraph_agent(self):
        from local_deep_research.advanced_search_system.strategies.langgraph_agent_strategy import (
            LangGraphAgentStrategy,
        )
        from local_deep_research.search_system_factory import create_strategy

        strategy = create_strategy(
            strategy_name="langgraph-agent",
            model=MagicMock(),
            search=MagicMock(),
            settings_snapshot={},
        )
        assert isinstance(strategy, LangGraphAgentStrategy)

    def test_factory_underscore_alias(self):
        from local_deep_research.advanced_search_system.strategies.langgraph_agent_strategy import (
            LangGraphAgentStrategy,
        )
        from local_deep_research.search_system_factory import create_strategy

        strategy = create_strategy(
            strategy_name="langgraph_agent",
            model=MagicMock(),
            search=MagicMock(),
            settings_snapshot={},
        )
        assert isinstance(strategy, LangGraphAgentStrategy)

    def test_strategy_in_available_list(self):
        from local_deep_research.search_system_factory import (
            get_available_strategies,
        )

        names = [s["name"] for s in get_available_strategies()]
        assert "langgraph-agent" in names

    def test_factory_passes_custom_params(self):
        from local_deep_research.search_system_factory import create_strategy

        strategy = create_strategy(
            strategy_name="langgraph-agent",
            model=MagicMock(),
            search=MagicMock(),
            settings_snapshot={},
            max_iterations=20,
            max_sub_iterations=3,
            include_sub_research=False,
        )
        assert strategy.max_iterations == 20
        assert strategy.max_sub_iterations == 3
        assert strategy.include_sub_research is False


# ---------------------------------------------------------------------------
# fetch_content collector registration (regression for PR #3457)
# ---------------------------------------------------------------------------


class TestFetchContentCollectorRegistration:
    """Regression coverage for PR #3457.

    Prior to the fix, ``_make_fetch_content_tool`` accepted ``collector`` but
    never used it, so every URL opened via the LLM's ``fetch_content`` tool
    was silently dropped from the final Sources section and citation system.
    These tests pin the fix: a successful fetch must register the URL, a
    duplicate fetch must reuse the existing citation index, and a failed
    fetch must not register anything.
    """

    def _make_collector(self):
        from local_deep_research.advanced_search_system.strategies.langgraph_agent_strategy import (
            SearchResultsCollector,
        )

        return SearchResultsCollector([])

    def _fetcher_cm(
        self, *, status="success", title="Page", content="Body", error=None
    ):
        """Return a MagicMock that behaves like ``ContentFetcher(...)``."""
        result = {"status": status, "title": title, "content": content}
        if error is not None:
            result["error"] = error
        fetcher = MagicMock()
        fetcher.fetch.return_value = result
        cm = MagicMock()
        cm.__enter__.return_value = fetcher
        cm.__exit__.return_value = False
        return cm

    def _make_tool(self, collector):
        from local_deep_research.advanced_search_system.tools.fetch import (
            build_fetch_tool,
        )

        return build_fetch_tool("full", collector)

    def test_successful_fetch_registers_url_in_collector(self):
        collector = self._make_collector()
        tool = self._make_tool(collector)
        cm = self._fetcher_cm(title="Hello", content="some body text")

        with patch(
            "local_deep_research.content_fetcher.ContentFetcher",
            return_value=cm,
        ):
            output = tool.invoke({"url": "http://example.com/page"})

        assert "http://example.com/page" in collector.sources
        assert len(collector.results) == 1
        entry = collector.results[0]
        assert entry["link"] == "http://example.com/page"
        assert entry["title"] == "Hello"
        assert entry["source_engine"] == "fetch"
        # Tool return is prefixed with the 1-based citation index so the
        # agent can cite fetched pages the same way it cites web_search hits.
        assert output.startswith("[1] ")

    def test_repeated_fetch_of_same_url_reuses_citation_index(self):
        collector = self._make_collector()
        # Simulate web_search having already captured this URL.
        collector.add_results(
            [
                {
                    "title": "From search",
                    "link": "http://example.com/page",
                    "snippet": "snip",
                }
            ],
            engine_name="web",
        )
        assert len(collector.results) == 1

        tool = self._make_tool(collector)
        cm = self._fetcher_cm(title="From fetch", content="full body")

        with patch(
            "local_deep_research.content_fetcher.ContentFetcher",
            return_value=cm,
        ):
            output = tool.invoke({"url": "http://example.com/page"})

        # No duplicate entry; the fetch reuses the existing citation slot.
        assert len(collector.results) == 1
        assert output.startswith("[1] ")

    def test_failed_fetch_does_not_register_url(self):
        collector = self._make_collector()
        tool = self._make_tool(collector)
        cm = self._fetcher_cm(
            status="error", title="", content="", error="timeout"
        )

        with patch(
            "local_deep_research.content_fetcher.ContentFetcher",
            return_value=cm,
        ):
            output = tool.invoke({"url": "http://broken.example/page"})

        assert collector.results == []
        assert collector.sources == []
        assert "Failed to fetch" in output

    def test_long_content_snippet_is_truncated_with_ellipsis(self):
        collector = self._make_collector()
        tool = self._make_tool(collector)
        cm = self._fetcher_cm(title="Long", content="A" * 500)

        with patch(
            "local_deep_research.content_fetcher.ContentFetcher",
            return_value=cm,
        ):
            tool.invoke({"url": "http://example.com/long"})

        snippet = collector.results[0]["snippet"]
        assert snippet.endswith("...")
        assert len(snippet) == 203  # 200 chars + "..."

    def test_find_by_url_returns_index_when_present(self):
        collector = self._make_collector()
        collector.add_results(
            [{"title": "A", "link": "http://a.com", "snippet": "a"}],
            engine_name="web",
        )
        assert collector.find_by_url("http://a.com") == 1

    def test_find_by_url_returns_none_when_absent(self):
        collector = self._make_collector()
        collector.add_results(
            [{"title": "A", "link": "http://a.com", "snippet": "a"}],
            engine_name="web",
        )
        assert collector.find_by_url("http://missing.com") is None


class TestFetchModeSettingResolution:
    """``LangGraphAgentStrategy.__init__`` reads the ``search.fetch.mode``
    setting (added in #3680; default changed to ``summary_focus_query``
    in #3793) and feeds it to ``build_fetch_tool``. The constructor must:

    - Accept any value in ``FETCH_MODES`` verbatim.
    - Reject any other value, log a warning, and fall back to
      ``summary_focus_query`` rather than crashing or letting an unknown
      mode reach ``build_fetch_tool``.

    The existing tests covered the constructor and tool-building paths
    but not this guard.
    """

    def _make_strategy(self, **overrides):
        from local_deep_research.advanced_search_system.strategies.langgraph_agent_strategy import (
            LangGraphAgentStrategy,
        )

        defaults = {
            "model": MagicMock(),
            "search": MagicMock(),
            "all_links_of_system": [],
            "settings_snapshot": {"search.tool": "duckduckgo"},
        }
        defaults.update(overrides)
        return LangGraphAgentStrategy(**defaults)

    def test_known_fetch_mode_accepted_verbatim(self):
        """``summary_focus`` (one of the ``FETCH_MODES``) must round-trip
        through the constructor unchanged.
        """
        strategy = self._make_strategy(
            settings_snapshot={
                "search.tool": "duckduckgo",
                "search.fetch.mode": "summary_focus",
            }
        )
        assert strategy.fetch_mode == "summary_focus"

    def test_unknown_fetch_mode_falls_back_to_default_with_warning(
        self, loguru_caplog
    ):
        """A misconfigured setting must not crash the constructor or
        propagate an unknown mode into ``build_fetch_tool``. The guard
        at the top of ``__init__`` logs a warning and substitutes the
        default. Anyone removing the guard would surface as the mode
        leaking through unchanged AND the warning going missing.
        """
        with loguru_caplog.at_level("WARNING"):
            strategy = self._make_strategy(
                settings_snapshot={
                    "search.tool": "duckduckgo",
                    "search.fetch.mode": "definitely-not-a-real-mode",
                }
            )

        assert strategy.fetch_mode == "summary_focus_query"
        assert "Unknown search.fetch.mode" in loguru_caplog.text
        assert "definitely-not-a-real-mode" in loguru_caplog.text

    def test_disabled_fetch_mode_omits_fetch_tool(self):
        """``fetch_mode='disabled'`` must produce a tool list with NO
        fetch tool — ``build_fetch_tool`` returns ``None`` and the
        ``if fetch is not None`` guard skips the append. A regression
        that always-appended would surface here as an extra tool.
        """
        strategy = self._make_strategy(
            settings_snapshot={
                "search.tool": "duckduckgo",
                "search.fetch.mode": "disabled",
            }
        )

        tools = strategy._build_tools(overall_query="anything")

        tool_names = {
            getattr(t, "name", None) or getattr(t, "__name__", None)
            for t in tools
        }
        # No tool whose name contains 'fetch'.
        assert all(
            "fetch" not in (name or "").lower() for name in tool_names
        ), (
            f"Expected no fetch tool with fetch_mode='disabled' but got "
            f"tools: {tool_names}"
        )


class TestResolveEngineNameIgnoresNonString:
    """``_resolve_engine_name`` short-circuits to the settings value only
    when it is a string (``isinstance(tool_setting, str)``); anything
    else — a list, a dict without a ``value`` key, an int — falls
    through to the class-name heuristic. The existing tests covered
    the success path and the bare-class fallback but didn't pin the
    non-string guard against realistic misconfiguration shapes.
    """

    def _make_strategy_with_search_tool_value(self, search_tool_value):
        from local_deep_research.advanced_search_system.strategies.langgraph_agent_strategy import (
            LangGraphAgentStrategy,
        )

        mock_search = MagicMock()
        mock_search.__class__.__name__ = "BraveSearchEngine"
        return LangGraphAgentStrategy(
            model=MagicMock(),
            search=mock_search,
            all_links_of_system=[],
            settings_snapshot={"search.tool": search_tool_value},
        )

    def test_list_settings_value_falls_through_to_class_heuristic(self):
        """A list at ``search.tool`` is not a valid engine name — the
        ``isinstance(..., str)`` guard rejects it and the class-name
        heuristic kicks in.
        """
        strategy = self._make_strategy_with_search_tool_value(
            ["this is not a string"]
        )
        assert strategy._search_engine_name == "brave"

    def test_int_settings_value_falls_through_to_class_heuristic(self):
        """Numeric values likewise fall through — pins that the guard
        rejects any non-string type, not just dicts.
        """
        strategy = self._make_strategy_with_search_tool_value(42)
        assert strategy._search_engine_name == "brave"


# ---------------------------------------------------------------------------
# Original research question must survive the tool-call display loop
# ---------------------------------------------------------------------------


class TestQueryParameterNotClobbered:
    """Regression for the ``query`` parameter clobber in ``analyze_topic``.

    The tool-call display loop builds a short label from each search tool's
    argument. A prior version assigned that label to ``query`` — the method
    parameter holding the *user's original research question* — so after the
    first ``web_search`` call, the original question was silently replaced by
    a truncated (<=80 char) search arg. That clobbered value then flowed into
    ``_finalize`` (the citation re-synthesis and the recorded
    ``findings[0]["question"]``) and the fallback ``_synthesize_from_collector``
    prompt, steering the final answer at the *wrong* question on the default
    research strategy. This test pins that the original question reaches
    ``_finalize`` unchanged after a run that issues a search tool call.
    """

    def _make_strategy(self):
        from local_deep_research.advanced_search_system.strategies.langgraph_agent_strategy import (
            LangGraphAgentStrategy,
        )

        return LangGraphAgentStrategy(
            model=MagicMock(),
            search=MagicMock(),
            all_links_of_system=[],
            settings_snapshot={"search.tool": {"value": "mock"}},
        )

    def test_original_query_reaches_finalize_after_search_tool_call(self):
        from langchain_core.messages import AIMessage

        strategy = self._make_strategy()

        original_query = (
            "What are the long-term cardiovascular effects of chronic sleep "
            "deprivation in adults over the age of fifty?"
        )

        # Agent emits a web_search tool call (whose arg differs from and is
        # shorter-after-truncation than the original question), then a final
        # answer message with no tool calls.
        tool_call_msg = AIMessage(
            content="",
            tool_calls=[
                {
                    "name": "web_search",
                    "args": {
                        "query": "sleep deprivation heart disease older adults"
                    },
                    "id": "call_1",
                    "type": "tool_call",
                }
            ],
        )
        answer_msg = AIMessage(content="Final synthesized answer with [1].")

        mock_agent = MagicMock()
        mock_agent.stream.return_value = iter(
            [
                {"agent": {"messages": [tool_call_msg]}},
                {"agent": {"messages": [answer_msg]}},
            ]
        )

        captured = {}

        def fake_finalize(query, final_answer, *args, **kwargs):
            captured["query"] = query
            return {
                "findings": [{"question": query, "content": final_answer}],
                "current_knowledge": final_answer,
                "iterations": 1,
                "error": None,
            }

        with (
            patch.object(strategy, "_build_tools", return_value=[MagicMock()]),
            patch("langchain.agents.create_agent", return_value=mock_agent),
            patch.object(strategy, "_update_progress"),
            patch.object(strategy, "_finalize", side_effect=fake_finalize),
        ):
            result = strategy.analyze_topic(original_query)

        # The user's original question — not the truncated search arg — must
        # reach _finalize and be recorded as the question.
        assert captured["query"] == original_query
        assert result["findings"][0]["question"] == original_query


class TestProgressMetadataKeepsStableId:
    """Progress metadata ``tool`` must carry the STABLE tool id while the
    human-readable engine label appears only in the message text.

    A prior revision of this PR overwrote ``metadata["tool"]`` with the
    friendly label; that discards the only machine-readable id reaching
    progress consumers. This pins the id-in-metadata / label-in-message
    split so a regression can't silently re-introduce the overwrite.
    """

    def _make_strategy(self):
        from local_deep_research.advanced_search_system.strategies.langgraph_agent_strategy import (
            LangGraphAgentStrategy,
        )

        return LangGraphAgentStrategy(
            model=MagicMock(),
            search=MagicMock(),
            all_links_of_system=[],
            settings_snapshot={"search.tool": {"value": "duckduckgo"}},
        )

    def test_tool_call_metadata_keeps_id_label_in_message(self):
        from langchain_core.messages import AIMessage

        strategy = self._make_strategy()

        tool_call_msg = AIMessage(
            content="",
            tool_calls=[
                {
                    "name": "web_search",
                    "args": {"query": "anything"},
                    "id": "call_1",
                    "type": "tool_call",
                }
            ],
        )
        answer_msg = AIMessage(content="Final answer with [1].")

        mock_agent = MagicMock()
        mock_agent.stream.return_value = iter(
            [
                {"agent": {"messages": [tool_call_msg]}},
                {"agent": {"messages": [answer_msg]}},
            ]
        )

        progress_calls = []

        def capture(*args, **kwargs):
            message = args[0] if args else kwargs.get("message", "")
            metadata = (
                args[2] if len(args) > 2 else kwargs.get("metadata", {})
            ) or {}
            progress_calls.append((message, metadata))

        with (
            patch.object(strategy, "_build_tools", return_value=[MagicMock()]),
            patch("langchain.agents.create_agent", return_value=mock_agent),
            patch.object(strategy, "_update_progress", side_effect=capture),
            patch.object(
                strategy,
                "_finalize",
                return_value={
                    "findings": [],
                    "current_knowledge": "",
                    "iterations": 1,
                    "error": None,
                },
            ),
        ):
            strategy.analyze_topic("test query")

        tool_calls = [
            (msg, md)
            for msg, md in progress_calls
            if md.get("phase") == "tool_call"
        ]
        assert tool_calls, "expected a tool_call progress event"
        message, metadata = tool_calls[0]
        # metadata keeps the stable id ...
        assert metadata["tool"] == "web_search"
        # ... while the user sees the brand label in the message text.
        assert "DuckDuckGo" in message


# ---------------------------------------------------------------------------
# Egress-scope tool filtering
# ---------------------------------------------------------------------------
#
# The strategy's ``_build_tools`` filters the specialized-engine tool list
# against the user's ``policy.egress_scope`` BEFORE the tools reach
# ``create_agent`` (see langgraph_agent_strategy.py line 591-655). That
# pre-filter is the "core fix for the original LangGraph silent-expansion
# complaint": the factory PEP would already refuse to instantiate a
# forbidden engine at runtime, but a runtime refusal still leaks policy
# state through the LLM's tool schema and through differential denial
# latency. Filtering the *list* means the forbidden tool names never
# enter the prompt at all.
#
# These tests pin that filter at the boundary that matters — the
# LangGraph tool list — using the real ``evaluate_engine`` /
# ``evaluate_retriever`` PDPs against a controlled engine fixture. A
# regression in either the strategy's filter loop OR the PDP itself
# shows up here.


class TestEgressScopeFiltering:
    """LangGraph tool list must honour ``policy.egress_scope`` so the LLM
    never even sees engines outside the active scope.
    """

    # Available-engines fixture. ``arxiv`` and ``pubmed`` are registered
    # public engines (``is_public = True`` on their classes); ``library``
    # is hardcoded local in ``evaluate_engine`` (line 322-326).
    # ``duckduckgo`` is the current primary — already added as
    # ``web_search`` and explicitly skipped at line 618.
    _FIXTURE_AVAILABLE = {
        "arxiv": {
            "is_local": False,
            "description": "arXiv preprints",
            "strengths": ["physics", "math"],
        },
        "pubmed": {
            "is_local": False,
            "description": "PubMed biomedical literature",
            "strengths": ["medicine"],
        },
        "library": {
            "is_local": True,
            "is_retriever": False,
            "description": "Local library",
            "strengths": ["personal documents"],
        },
        # A per-collection engine. evaluate_engine hardcodes the
        # ``collection_*`` name prefix as local (egress_policy.py ~322),
        # a DISTINCT code path from the ``library`` all-collections engine.
        "collection_abc123": {
            "is_local": True,
            "is_retriever": False,
            "description": "My research papers (Collection)",
            "strengths": ["curated documents"],
        },
        "duckduckgo": {
            "is_local": False,
            "description": "DuckDuckGo",
        },
    }

    def _make_strategy(self, scope, primary_engine="duckduckgo"):
        from local_deep_research.advanced_search_system.strategies.langgraph_agent_strategy import (
            LangGraphAgentStrategy,
        )

        mock_search = MagicMock()
        mock_search.__class__.__name__ = "DuckDuckGoSearchEngine"
        return LangGraphAgentStrategy(
            model=MagicMock(),
            search=mock_search,
            all_links_of_system=[],
            settings_snapshot={
                "search.tool": primary_engine,
                "policy.egress_scope": scope,
            },
        )

    @staticmethod
    def _tool_names(tools):
        names = set()
        for t in tools:
            name = getattr(t, "name", None) or getattr(t, "__name__", None)
            if name:
                names.add(name)
        return names

    # ------------------------------------------------------------------
    # STRICT — only the primary web_search; NO specialized engines at all
    # ------------------------------------------------------------------

    def test_collection_engine_treated_as_local(self):
        """A per-collection ``collection_<id>`` engine hits a DISTINCT
        classifier branch from ``library`` (the name-prefix rule in
        evaluate_engine, not a config flag). Pin that it behaves as local:
        present under PRIVATE_ONLY, filtered under PUBLIC_ONLY.
        """
        # PRIVATE_ONLY: the collection survives (it's local).
        strat_priv = self._make_strategy(scope="private_only")
        with patch(
            "local_deep_research.web_search_engines.search_engines_config.get_available_engines",
            return_value=self._FIXTURE_AVAILABLE,
        ):
            priv_names = self._tool_names(
                strat_priv._build_tools(overall_query="q")
            )
        assert "search_collection_abc123" in priv_names, (
            "collection_<id> is local — must pass PRIVATE_ONLY"
        )

        # PUBLIC_ONLY: the collection is filtered (local data stays local).
        strat_pub = self._make_strategy(scope="public_only")
        with patch(
            "local_deep_research.web_search_engines.search_engines_config.get_available_engines",
            return_value=self._FIXTURE_AVAILABLE,
        ):
            pub_names = self._tool_names(
                strat_pub._build_tools(overall_query="q")
            )
        assert "search_collection_abc123" not in pub_names, (
            "collection_<id> is local — must be filtered under PUBLIC_ONLY"
        )

    def test_strict_registers_no_specialized_search_tools(self):
        """STRICT means the agent gets only the primary ``web_search``
        (plus generic helpers like fetch_content / research_subtopic).
        Every ``search_*`` tool — public OR local — must be filtered
        out by the ``continue`` at line 623-627.
        """
        strategy = self._make_strategy(scope="strict")
        with patch(
            "local_deep_research.web_search_engines.search_engines_config.get_available_engines",
            return_value=self._FIXTURE_AVAILABLE,
        ):
            tools = strategy._build_tools(overall_query="q")

        names = self._tool_names(tools)
        # The primary web_search is unaffected.
        assert "web_search" in names
        # No specialized search_* — not arxiv, not pubmed, not library.
        specialized = {n for n in names if n.startswith("search_")}
        assert specialized == set(), (
            f"STRICT must register zero specialized search_* tools, "
            f"got: {specialized}"
        )

    # ------------------------------------------------------------------
    # PRIVATE_ONLY — public engines filtered, local engines kept
    # ------------------------------------------------------------------

    def test_private_only_filters_out_public_specialized_engines(self):
        """Under PRIVATE_ONLY the agent must NOT see arxiv or pubmed —
        ``scope_mismatch_private_only`` from ``evaluate_engine`` — but
        library (``is_local=True``) passes through.
        """
        strategy = self._make_strategy(scope="private_only")
        with patch(
            "local_deep_research.web_search_engines.search_engines_config.get_available_engines",
            return_value=self._FIXTURE_AVAILABLE,
        ):
            tools = strategy._build_tools(overall_query="q")

        names = self._tool_names(tools)
        assert "search_arxiv" not in names, (
            "arXiv is public — must be filtered under PRIVATE_ONLY"
        )
        assert "search_pubmed" not in names, (
            "PubMed is public — must be filtered under PRIVATE_ONLY"
        )
        assert "search_library" in names, (
            "library is local — must pass PRIVATE_ONLY filter"
        )

    # ------------------------------------------------------------------
    # PUBLIC_ONLY — local engines filtered, public engines kept
    # ------------------------------------------------------------------

    def test_public_only_filters_out_local_specialized_engines(self):
        """Under PUBLIC_ONLY the agent must NOT see ``search_library`` —
        ``scope_mismatch_public_only`` — but arxiv and pubmed remain.
        This is the user-data-stays-on-the-box property: a PUBLIC_ONLY
        run must never load local indexes into the agent's tool surface.
        """
        strategy = self._make_strategy(scope="public_only")
        with patch(
            "local_deep_research.web_search_engines.search_engines_config.get_available_engines",
            return_value=self._FIXTURE_AVAILABLE,
        ):
            tools = strategy._build_tools(overall_query="q")

        names = self._tool_names(tools)
        assert "search_library" not in names, (
            "library is local — must be filtered under PUBLIC_ONLY"
        )
        assert "search_arxiv" in names, (
            "arXiv is public — must pass PUBLIC_ONLY filter"
        )
        assert "search_pubmed" in names, (
            "PubMed is public — must pass PUBLIC_ONLY filter"
        )

    # ------------------------------------------------------------------
    # BOTH (default) — every classified engine is registered
    # ------------------------------------------------------------------

    def test_both_scope_registers_every_classified_engine(self):
        """The default scope BOTH must register every classified engine
        in the available dict. The current primary is excluded by the
        explicit ``continue`` at line 618 — NOT by the scope filter — so
        a regression that moved it into the scope-mismatch path would
        still be caught by the assertion that it's absent.
        """
        strategy = self._make_strategy(scope="both")
        with patch(
            "local_deep_research.web_search_engines.search_engines_config.get_available_engines",
            return_value=self._FIXTURE_AVAILABLE,
        ):
            tools = strategy._build_tools(overall_query="q")

        names = self._tool_names(tools)
        for expected in ("search_arxiv", "search_pubmed", "search_library"):
            assert expected in names, (
                f"Expected {expected} under BOTH but got: {sorted(names)}"
            )
        # The current engine is NEVER added as a specialized tool
        # regardless of scope.
        assert "search_duckduckgo" not in names

    # ------------------------------------------------------------------
    # Fail-closed: corrupted scope value
    # ------------------------------------------------------------------

    def test_corrupted_scope_value_propagates_policy_denied(self):
        """A junk ``policy.egress_scope`` value must NOT silently fall
        through to BOTH (the most permissive scope). ``context_from_snapshot``
        raises ``PolicyDeniedError(unknown_egress_scope)``; the strategy's
        ``_build_egress_context`` re-raises it (only ValueError / KeyError /
        TypeError get swallowed). The run aborts instead of running
        unfiltered.
        """
        from local_deep_research.security.egress.policy import (
            PolicyDeniedError,
        )

        strategy = self._make_strategy(scope="not-a-real-scope")
        with pytest.raises(PolicyDeniedError):
            strategy._build_tools(overall_query="q")

    # ------------------------------------------------------------------
    # Audit log — every block must leave an audit-bound trail
    # ------------------------------------------------------------------

    def test_blocked_engine_emits_policy_audit_log(self, loguru_caplog):
        """When the filter drops an engine, the strategy emits the
        ``specialized tool filtered by egress policy`` info line. Under
        PUBLIC_ONLY with this fixture exactly one engine (``library``)
        is local, so the line must fire exactly once — a regression
        that bypassed the filter would fire zero times, and a regression
        that over-filtered (e.g. also dropped public engines under
        PUBLIC_ONLY) would fire more than once.

        Note: ``logger.bind(policy_audit=True).info("...", engine=..., ...)``
        attaches the engine name and the ``policy_audit`` flag as loguru
        record extras, NOT to the rendered message text. Asserting the
        bound flag itself would require a custom loguru sink; we settle
        for the rendered-line invariant here.
        """
        strategy = self._make_strategy(scope="public_only")
        with (
            loguru_caplog.at_level("INFO"),
            patch(
                "local_deep_research.web_search_engines.search_engines_config.get_available_engines",
                return_value=self._FIXTURE_AVAILABLE,
            ),
        ):
            strategy._build_tools(overall_query="q")

        marker = "specialized tool filtered by egress policy"
        occurrences = loguru_caplog.text.count(marker)
        # Under PUBLIC_ONLY every LOCAL engine in the fixture is dropped:
        # ``library`` and ``collection_abc123``. One audit line per drop.
        local_engine_count = sum(
            1
            for name, cfg in self._FIXTURE_AVAILABLE.items()
            if cfg.get("is_local") is True
        )
        assert occurrences == local_engine_count, (
            f"Expected one audit-log line per dropped local engine "
            f"({local_engine_count}), got {occurrences}. Captured text:\n"
            f"{loguru_caplog.text}"
        )


# ---------------------------------------------------------------------------
# Policy addendum — the LLM-facing scope signal
# ---------------------------------------------------------------------------
#
# Filtering the tool LIST closes the latency-leak half of the timing
# attack. The other half is the prompt addendum: the LLM is *told* which
# tools exist so it doesn't waste tokens probing for forbidden engines.
# These tests pin that the addendum text varies by scope and is empty
# under BOTH (we don't want to bleed policy state into the LLM for the
# default scope).


class TestEgressScopePolicyAddendum:
    """``analyze_topic`` injects a policy addendum into the system prompt
    that gets passed to ``create_agent``. The addendum's presence and
    wording must reflect the active scope.
    """

    def _make_strategy(self, scope, primary_engine="duckduckgo"):
        from local_deep_research.advanced_search_system.strategies.langgraph_agent_strategy import (
            LangGraphAgentStrategy,
        )

        mock_search = MagicMock()
        mock_search.__class__.__name__ = "DuckDuckGoSearchEngine"
        return LangGraphAgentStrategy(
            model=MagicMock(),
            search=mock_search,
            all_links_of_system=[],
            settings_snapshot={
                "search.tool": primary_engine,
                "policy.egress_scope": scope,
            },
        )

    def _capture_prompt(self, scope, primary="duckduckgo"):
        """Run analyze_topic in a heavily-mocked harness and return the
        ``system_prompt`` string passed to ``create_agent``. There's no
        smaller public hook for the addendum — the prompt-string is
        the surface the LLM actually receives.
        """
        from langchain_core.messages import AIMessage

        strategy = self._make_strategy(scope=scope, primary_engine=primary)
        captured = {}

        mock_agent = MagicMock()
        mock_agent.stream.return_value = iter(
            [{"agent": {"messages": [AIMessage(content="done")]}}]
        )

        def fake_create_agent(model=None, tools=None, system_prompt=None, **kw):
            captured["system_prompt"] = system_prompt
            return mock_agent

        with (
            patch.object(strategy, "_build_tools", return_value=[MagicMock()]),
            patch(
                "langchain.agents.create_agent",
                side_effect=fake_create_agent,
            ),
            patch.object(strategy, "_update_progress"),
            patch.object(
                strategy,
                "_finalize",
                return_value={
                    "findings": [],
                    "current_knowledge": "",
                    "iterations": 0,
                    "error": None,
                },
            ),
        ):
            strategy.analyze_topic("q")
        return captured.get("system_prompt", "") or ""

    def test_strict_addendum_locks_llm_to_primary_engine(self):
        """STRICT must tell the LLM that ``search_*`` tools don't exist
        and name the primary engine — otherwise the LLM may probe for
        a denied tool, and the denial latency leaks policy state.
        """
        prompt = self._capture_prompt("strict")
        assert "RESTRICTED MODE" in prompt
        # The primary engine name must be cited.
        assert "duckduckgo" in prompt.lower()

    def test_private_only_addendum_names_public_engines_as_unavailable(self):
        """PRIVATE-ONLY addendum must explicitly warn the LLM that
        public engines are out of scope so it doesn't waste turns
        calling search_arxiv etc.
        """
        prompt = self._capture_prompt("private_only")
        assert "PRIVATE-ONLY MODE" in prompt
        # Names at least one canonical public engine so the LLM
        # generalises correctly.
        assert "arxiv" in prompt.lower()

    def test_public_only_addendum_names_local_engines_as_unavailable(self):
        """PUBLIC-ONLY addendum must mark local tools as unavailable —
        and it must NOT be the STRICT addendum (different scope, different
        rules).
        """
        prompt = self._capture_prompt("public_only")
        assert "PUBLIC-ONLY MODE" in prompt
        assert "RESTRICTED MODE" not in prompt
        # Names at least one canonical local tool.
        assert "library" in prompt.lower()

    def test_both_scope_injects_no_policy_addendum(self):
        """Under BOTH (default), the strategy MUST NOT inject any of the
        three scope-specific marker phrases. Bleeding scope state into
        every prompt would (a) bloat the default-case prompt for no
        reason and (b) leak which scope the user picked even when they
        didn't restrict anything.
        """
        prompt = self._capture_prompt("both")
        assert "RESTRICTED MODE" not in prompt
        assert "PRIVATE-ONLY MODE" not in prompt
        assert "PUBLIC-ONLY MODE" not in prompt
