learningcircuit--local-deep-research/tests/web/services/test_research_service_progress_integration.py

"""Integration tests against the ACTUAL progress_callback closure inside
run_research_process — not a replicated helper.

Why this file exists
--------------------
The unit tests in ``test_research_service_execution.py`` exercise the
detailed-mode progress logic via a local ``_apply_detailed_progress``
helper that mirrors production. That pattern is enough for arithmetic
edge cases, but it cannot catch bugs in the integration boundary
between the closure, ``SearchSystem.set_progress_callback``, and
``globals.update_progress_and_check_active``.

PR #3806's original review missed exactly such an integration bug
(F1): the closure pinned the bar to 100 whenever ``phase="complete"``
arrived — including the mid-report emissions that every strategy
produces from inside ``analyze_topic``. The replica-helper tests
passed; an integration test would have caught it.

Pattern
-------
1. Mock ``AdvancedSearchSystem`` with a ``side_effect`` on
   ``set_progress_callback`` that captures the inner closure.
2. Make ``analyze_topic`` raise ``ResearchTerminatedException`` so
   ``run_research_process`` exits gracefully — but the callback has
   been captured.
3. Mock ``update_progress_and_check_active`` with a stateful
   ``side_effect`` that mimics ``globals.py``'s monotonic guard.
4. Yield the captured callback while patches are still active so
   tests can call it with bug-triggering sequences.
"""

from contextlib import contextmanager
from unittest.mock import MagicMock, Mock, patch

import pytest
from flask import Flask
from loguru import logger

# The production callback emits at the custom "MILESTONE" log level (registered
# by log_utils.init_loguru). Tests don't run init_loguru, so we register it
# here once. Idempotent if it already exists.
try:
    logger.level("MILESTONE", no=26)
except (ValueError, TypeError):
    pass


@pytest.fixture(scope="module")
def flask_app():
    """Flask app context for @log_for_research-decorated run_research_process."""
    app = Flask(__name__)
    app.config["TESTING"] = True
    return app


@contextmanager
def _noop_db_session(*_, **__):
    session = MagicMock()
    session.__enter__ = Mock(return_value=session)
    session.__exit__ = Mock(return_value=False)
    yield session


@contextmanager
def captured_progress_callback(mode, flask_app, run_kwargs=None, extras=None):
    """Yield (callback, progress_state, update_calls) with patches active.

    progress_state[0] is the value globals.update_progress_and_check_active
    would have stored. update_calls is the list of (rid, new, stored) for
    every invocation, so tests can assert on order and rejection of lower
    values.

    ``run_kwargs`` overrides/extends the ``run_research_process`` call
    (e.g. ``chat_session_id`` to exercise the chat-step path, or a unique
    ``research_id`` to isolate the module-global emit-throttle state).
    ``extras``, when given a dict, is populated with the ``socketio``
    service mock and the patched ``chat_service_cls`` so tests can assert
    on emits and persisted steps.
    """
    from local_deep_research.exceptions import ResearchTerminatedException
    from local_deep_research.web.services.research_service import (
        run_research_process,
    )

    captured = {}

    def capture(cb):
        captured["callback"] = cb

    mock_system = Mock()
    mock_system.set_progress_callback = Mock(side_effect=capture)
    mock_system.analyze_topic.side_effect = ResearchTerminatedException(
        "captured for test"
    )

    progress_state = [0]
    update_calls = []

    def fake_monotonic_update(rid, new_progress):
        if new_progress is not None and new_progress > progress_state[0]:
            progress_state[0] = new_progress
        update_calls.append((rid, new_progress, progress_state[0]))
        return (progress_state[0], True)

    socketio_service = Mock()
    chat_service_cls = MagicMock()
    if extras is not None:
        extras["socketio"] = socketio_service
        extras["chat_service_cls"] = chat_service_cls

    with flask_app.app_context():
        with (
            patch(
                "local_deep_research.web.routes.globals.is_termination_requested",
                return_value=False,
            ),
            patch(
                "local_deep_research.web.routes.globals.is_research_active",
                return_value=True,
            ),
            # run_research_process imports update_progress_and_check_active
            # from ..routes.globals at function scope (L319), so we patch the
            # source rather than the consuming module.
            patch(
                "local_deep_research.web.routes.globals.update_progress_and_check_active",
                side_effect=fake_monotonic_update,
            ),
            patch(
                "local_deep_research.web.services.research_service.cleanup_research_resources"
            ),
            patch(
                "local_deep_research.web.services.research_service.get_user_db_session",
                side_effect=_noop_db_session,
            ),
            patch(
                "local_deep_research.web.services.research_service.AdvancedSearchSystem",
                return_value=mock_system,
            ),
            patch(
                "local_deep_research.web.services.research_service.set_search_context"
            ),
            patch(
                "local_deep_research.config.thread_settings.set_settings_context"
            ),
            patch("local_deep_research.settings.logger.log_settings"),
            patch(
                "local_deep_research.web.services.research_service.SocketIOService",
                return_value=socketio_service,
            ),
            # The chat-step persist path imports ChatService at function
            # scope inside the callback — patch the source module.
            patch(
                "local_deep_research.chat.service.ChatService",
                chat_service_cls,
            ),
            patch("local_deep_research.web.queue.processor_v2.queue_processor"),
            patch(
                "local_deep_research.web.services.research_service.handle_termination"
            ),
        ):
            call_kwargs = dict(
                research_id=1,
                query="test query",
                mode=mode,
                username="testuser",
                # A real run always has a configured primary; without it the
                # egress context build fails closed (resolve_run_primary_engine
                # raises) and the worker refuses the run before the callback
                # wiring under test runs.
                settings_snapshot={"search.tool": "searxng"},
            )
            if run_kwargs:
                call_kwargs.update(run_kwargs)
            run_research_process(**call_kwargs)
            yield captured.get("callback"), progress_state, update_calls


class TestRunResearchProcessProgressCallbackIntegration:
    """End-to-end behaviour of the closure wired through SearchSystem + globals."""

    def test_callback_is_captured_in_detailed_mode(self, flask_app):
        """Sanity: set_progress_callback receives a callable."""
        with captured_progress_callback("detailed", flask_app) as (
            cb,
            _,
            _,
        ):
            assert cb is not None, (
                "run_research_process did not call set_progress_callback "
                "before exiting"
            )
            assert callable(cb)

    def test_strategy_complete_does_not_pin_bar_to_100_mid_report(
        self, flask_app
    ):
        """Regression for F1 (PR #3806 review).

        Every strategy emits ``{"phase": "complete"}`` at the end of its
        ``analyze_topic``. ``report_generator._research_and_generate_sections``
        runs ``self.search_system.analyze_topic`` per subsection, and the
        SearchSystem's callback is the OUTER closure (set at
        research_service.py:711, not the wrapper). Pre-fix the closure had
        ``elif phase == "complete": adjusted_progress = 100``, which fired
        after every subsection and pinned the bar at 100 for the rest of
        a multi-minute report.

        Post-fix: ``phase="complete"`` is capped at the search cap (8) and
        the globals' monotonic guard preserves the existing higher value.
        """
        with captured_progress_callback("detailed", flask_app) as (
            cb,
            progress_state,
            _,
        ):
            # Baseline: bar climbs into the report range
            cb(
                "Generating detailed report...",
                10,
                {"phase": "report_generation"},
            )
            assert progress_state[0] == 10

            # Strategy completes its analyze_topic mid-report — this fires
            # every subsection, e.g. via standard_strategy.py:334 at value 95.
            cb("Research complete", 95, {"phase": "complete"})

            assert progress_state[0] == 10, (
                f"regression: bar jumped to {progress_state[0]} after mid-"
                "report strategy 'complete' emission. Expected 10 (the bar "
                "should stay at the report-phase value; 'complete' is capped "
                "at 8 and rejected by monotonic guard)."
            )

    def test_report_phase_sequence_climbs_through_closure(self, flask_app):
        """A normal report-phase emission sequence drives the bar 10 → 100."""
        with captured_progress_callback("detailed", flask_app) as (
            cb,
            progress_state,
            _,
        ):
            cb("structure", 10, {"phase": "report_generation"})
            cb("section 1", 30, {"phase": "report_section_research"})
            cb("section 2", 60, {"phase": "report_section_research"})
            cb("formatting", 91, {"phase": "report_formatting"})
            cb("complete", 100, {"phase": "report_complete"})
            assert progress_state[0] == 100

    def test_search_phase_capped_through_closure(self, flask_app):
        """Search-phase emissions are capped at the configured search cap."""
        from local_deep_research.web.services.research_service import (
            _DETAILED_SEARCH_PROGRESS_CAP,
        )

        with captured_progress_callback("detailed", flask_app) as (
            cb,
            progress_state,
            _,
        ):
            # Large raw values from search phase get capped.
            cb("search 1", 50, {"phase": "search"})
            assert progress_state[0] == _DETAILED_SEARCH_PROGRESS_CAP

            cb("search 2", 137, {"phase": "search"})
            assert progress_state[0] == _DETAILED_SEARCH_PROGRESS_CAP

    def test_none_progress_does_not_crash_or_pin(self, flask_app):
        """None progress (e.g. error path, sub-search relays) passes through.

        Regression for the None-guard fix from commit 5f4ae17ea: the new
        ``elif progress_percent is not None`` branch in detailed mode must
        not raise ``TypeError`` on ``min(8, None)``.
        """
        with captured_progress_callback("detailed", flask_app) as (
            cb,
            progress_state,
            _,
        ):
            cb("search 1", 5, {"phase": "search"})
            baseline = progress_state[0]

            # Should not raise; should not affect stored progress.
            cb("error", None, {"phase": "error"})
            cb("sub-search complete", None, {"phase": "search_complete"})

            assert progress_state[0] == baseline


class TestChatStepEmitBypassesThrottle:
    """Persisted chat steps must always emit (PR #5051).

    The 200ms socket-emit throttle used to drop back-to-back observation
    events (the agent issuing several tool calls in one turn) from the
    live stream even though every one was persisted as a chat step — the
    live UI showed fewer steps than a session reload reconstructed,
    violating the symmetry invariant in the live direction. The chat-step
    decision now runs before the throttle gate and forces the emit for
    any event that persists a step.
    """

    def _progress_emits(self, socketio):
        return [
            c
            for c in socketio.emit_to_subscribers.call_args_list
            if c.args[0] == "progress"
        ]

    def test_back_to_back_observations_all_emit_and_persist(self, flask_app):
        extras = {}
        with captured_progress_callback(
            "quick",
            flask_app,
            # Unique research_id isolates the module-global throttle state.
            run_kwargs={
                "research_id": 987654,
                "chat_session_id": "sess-throttle-test",
            },
            extras=extras,
        ) as (cb, _, __):
            socketio = extras["socketio"]
            add_step = extras["chat_service_cls"].return_value.add_progress_step
            socketio.emit_to_subscribers.reset_mock()
            add_step.reset_mock()

            detail_1 = "x" * 200
            detail_2 = "y" * 200
            cb(
                "📄 From the web (SearXNG): first",
                42,
                {
                    "phase": "observation",
                    "tool": "web_search",
                    "content": detail_1,
                },
            )
            # Fired microseconds later — inside the 200ms throttle window.
            cb(
                "📄 From the web (SearXNG): second",
                43,
                {
                    "phase": "observation",
                    "tool": "web_search",
                    "content": detail_2,
                },
            )

            emits = self._progress_emits(socketio)
            assert len(emits) == 2, (
                "second back-to-back observation was throttled off the "
                "live stream despite being persisted as a chat step"
            )
            # The detail rides along on the wire for the live step…
            assert emits[1].args[2]["content"] == detail_2
            # …and both events persisted the composed message + detail.
            assert add_step.call_count == 2
            assert add_step.call_args_list[1].kwargs["content"] == (
                "📄 From the web (SearXNG): second\n\n" + detail_2
            )

    def test_non_persisted_repeat_phase_still_suppressed(self, flask_app):
        """The forced emit is persist-gated: a deduped repeat step phase
        (persist=False, non-final) must still be suppressed even when the
        throttle would let it through."""
        import local_deep_research.web.services.research_service as rs

        extras = {}
        with captured_progress_callback(
            "quick",
            flask_app,
            run_kwargs={
                "research_id": 987655,
                "chat_session_id": "sess-suppress-test",
            },
            extras=extras,
        ) as (cb, _, __):
            socketio = extras["socketio"]
            add_step = extras["chat_service_cls"].return_value.add_progress_step
            socketio.emit_to_subscribers.reset_mock()
            add_step.reset_mock()

            # Disable the throttle so suppression is the ONLY thing that
            # can drop the second emit — otherwise the throttle window
            # would mask a broken suppress path.
            with patch.object(rs, "_EMIT_THROTTLE_SECONDS", 0.0):
                cb("Searching the web…", 20, {"phase": "search"})
                cb("Searching the web again…", 21, {"phase": "search"})

            assert len(self._progress_emits(socketio)) == 1, (
                "deduped repeat 'search' step leaked to the live stream — "
                "reload would not reconstruct it (symmetry violation)"
            )
            assert add_step.call_count == 1