learningcircuit--local-deep-research/tests/web/queue/test_processor_v2_concurrency.py

"""
Tests for QueueProcessorV2 error recovery, cleanup, concurrency, and
process_user_request — replacing fake tests that never import the real class.

Source: src/local_deep_research/web/queue/processor_v2.py
"""

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

import pytest

from local_deep_research.web.queue.processor_v2 import QueueProcessorV2


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------


@pytest.fixture
def processor():
    """Create a fresh QueueProcessorV2 instance (not started)."""
    return QueueProcessorV2(check_interval=1)


@contextmanager
def _mock_db_session():
    """Yield a chainable mock db session."""
    mock_session = Mock()
    mock_query = Mock()
    mock_query.filter_by.return_value = mock_query
    mock_query.filter.return_value = mock_query
    mock_query.order_by.return_value = mock_query
    mock_query.limit.return_value = mock_query
    mock_query.first.return_value = None
    mock_query.all.return_value = []
    mock_query.count.return_value = 0
    mock_session.query.return_value = mock_query
    yield mock_session


# ---------------------------------------------------------------------------
# _commit_with_safe_rollback / _delete_queue_row_safely helpers
# ---------------------------------------------------------------------------


class TestCommitWithSafeRollback:
    """The commit-with-safe-rollback helper is the base primitive used
    throughout _start_queued_researches; verify each branch in isolation."""

    def test_commit_success_returns_true_no_rollback(self, processor):
        mock_session = Mock()
        result = processor._commit_with_safe_rollback(mock_session, "ctx")
        assert result is True
        mock_session.commit.assert_called_once()
        mock_session.rollback.assert_not_called()

    def test_commit_failure_returns_false_and_rolls_back(self, processor):
        mock_session = Mock()
        mock_session.commit.side_effect = RuntimeError("commit boom")
        result = processor._commit_with_safe_rollback(mock_session, "ctx")
        assert result is False
        mock_session.commit.assert_called_once()
        mock_session.rollback.assert_called_once()

    def test_commit_and_rollback_both_raise_does_not_propagate(self, processor):
        """If both commit and rollback fail, the helper must NOT
        propagate — the caller has no way to recover and we are already
        in a best-effort cleanup path."""
        mock_session = Mock()
        mock_session.commit.side_effect = RuntimeError("commit boom")
        mock_session.rollback.side_effect = RuntimeError("rollback boom")
        result = processor._commit_with_safe_rollback(mock_session, "ctx")
        assert result is False


class TestDeleteQueueRowSafely:
    """The dup-branch cleanup helper: rollback-first, re-query, delete,
    commit. All failures absorbed — this is a best-effort path."""

    def _session_with_queued_row(self, queued_row):
        """Mock session where the QueuedResearch filter_by().first()
        returns ``queued_row`` (or None)."""
        mock_session = Mock()
        mock_query = Mock()
        mock_query.filter_by.return_value = mock_query
        mock_query.first.return_value = queued_row
        mock_session.query.return_value = mock_query
        return mock_session

    def test_deletes_existing_row_and_commits(self, processor):
        queued_row = Mock()
        mock_session = self._session_with_queued_row(queued_row)

        processor._delete_queue_row_safely(mock_session, "alice", "r-1")

        mock_session.delete.assert_called_once_with(queued_row)
        mock_session.commit.assert_called_once()

    def test_no_row_present_does_not_call_delete_but_still_commits(
        self, processor
    ):
        """If the row was already deleted by a concurrent worker / prior
        retry, the helper must not call ``delete()`` with ``None`` but
        still issue the commit to flush the preceding rollback."""
        mock_session = self._session_with_queued_row(None)

        processor._delete_queue_row_safely(mock_session, "alice", "r-1")

        mock_session.delete.assert_not_called()
        mock_session.commit.assert_called_once()

    def test_initial_rollback_failure_does_not_prevent_delete(self, processor):
        """The pre-emptive rollback is best-effort; if it fails we still
        try the query+delete+commit sequence."""
        queued_row = Mock()
        mock_session = self._session_with_queued_row(queued_row)
        mock_session.rollback.side_effect = RuntimeError("rollback boom")

        processor._delete_queue_row_safely(mock_session, "alice", "r-1")

        mock_session.delete.assert_called_once_with(queued_row)
        mock_session.commit.assert_called_once()

    def test_query_failure_is_absorbed(self, processor):
        """If the re-query itself raises (session truly poisoned), the
        helper must not propagate — the caller has no handler for this."""
        mock_session = Mock()
        mock_session.query.side_effect = RuntimeError("query boom")

        # No exception expected.
        processor._delete_queue_row_safely(mock_session, "alice", "r-1")


# ---------------------------------------------------------------------------
# _start_queued_researches error recovery
# ---------------------------------------------------------------------------


class TestStartQueuedResearchesErrorRecovery:
    """When _start_research throws, verify is_processing reset and task FAILED."""

    def test_error_resets_is_processing(self, processor):
        """On exception, is_processing must be set back to False so the
        next loop tick can retry."""
        queued = Mock()
        queued.is_processing = False
        queued.research_id = "res-1"

        mock_session = Mock()
        mock_query = Mock()
        mock_query.filter_by.return_value = mock_query
        mock_query.order_by.return_value = mock_query
        mock_query.limit.return_value = mock_query
        mock_query.all.return_value = [queued]
        # Error handler re-queries the row; return the same object.
        mock_query.first.return_value = queued
        mock_session.query.return_value = mock_query

        mock_queue_service = Mock()

        # _start_research throws
        processor._start_research = Mock(side_effect=RuntimeError("boom"))

        processor._start_queued_researches(
            mock_session, mock_queue_service, "alice", "pw", 1
        )

        assert queued.is_processing is False

    def test_transient_error_leaves_queued_for_retry(self, processor):
        """On a single spawn failure, the queue row stays and is_processing
        is reset, allowing the next loop tick to retry."""
        queued = Mock()
        queued.is_processing = False
        queued.research_id = "res-2"

        mock_session = Mock()
        mock_query = Mock()
        mock_query.filter_by.return_value = mock_query
        mock_query.order_by.return_value = mock_query
        mock_query.limit.return_value = mock_query
        mock_query.all.return_value = [queued]
        mock_query.first.return_value = queued
        mock_session.query.return_value = mock_query

        mock_queue_service = Mock()
        processor._start_research = Mock(side_effect=RuntimeError("boom"))

        processor._start_queued_researches(
            mock_session, mock_queue_service, "alice", "pw", 1
        )

        # Row must remain queued for retry — not deleted, not marked FAILED.
        mock_session.delete.assert_not_called()
        assert queued.is_processing is False
        assert processor._spawn_retry_counts["res-2"] == 1

    def test_terminal_failure_after_retry_limit(self, processor):
        """After SPAWN_RETRY_LIMIT consecutive failures, row is deleted,
        ResearchHistory.status is set to FAILED, and notify_research_failed
        is invoked exactly once."""
        from local_deep_research.constants import ResearchStatus
        from local_deep_research.database.models import (
            QueuedResearch,
            ResearchHistory,
        )
        from local_deep_research.web.queue.processor_v2 import (
            SPAWN_RETRY_LIMIT,
        )

        queued = Mock()
        queued.is_processing = False
        queued.research_id = "res-3"

        research_row = Mock()

        # Route queries by model class: QueuedResearch re-query returns
        # the queued row; ResearchHistory lookup returns research_row.
        def _make_query(model):
            q = Mock()
            q.filter_by.return_value = q
            q.order_by.return_value = q
            q.limit.return_value = q
            if model is QueuedResearch:
                q.all.return_value = [queued]
                q.first.return_value = queued
            elif model is ResearchHistory:
                q.first.return_value = research_row
            else:
                q.all.return_value = []
                q.first.return_value = None
            return q

        mock_session = Mock()
        mock_session.query.side_effect = _make_query

        mock_queue_service = Mock()
        processor._start_research = Mock(side_effect=RuntimeError("boom"))
        processor.notify_research_failed = Mock()

        # Simulate SPAWN_RETRY_LIMIT loop iterations: each call processes
        # the same queued row and fails.
        for _ in range(SPAWN_RETRY_LIMIT):
            processor._start_queued_researches(
                mock_session, mock_queue_service, "alice", "pw", 1
            )

        # After exhausting attempts: terminal path triggered once.
        assert "res-3" not in processor._spawn_retry_counts
        assert processor.notify_research_failed.call_count == 1
        call_args = processor.notify_research_failed.call_args
        assert call_args.kwargs["username"] == "alice"
        assert call_args.kwargs["research_id"] == "res-3"
        # Queue row deleted and ResearchHistory marked FAILED.
        mock_session.delete.assert_called_with(queued)
        assert research_row.status == ResearchStatus.FAILED

    def test_capacity_reject_does_not_count_toward_retry_limit(self, processor):
        """A SystemAtCapacityError re-raised from _start_research is a
        transient condition (system globally at capacity), NOT a spawn
        failure. It must NOT bump the spawn-retry counter or mark the
        research FAILED — otherwise a busy system destroys perfectly valid
        queued work after a few ticks. Regression for the missing dedicated
        `except SystemAtCapacityError` clause in _start_queued_researches."""
        from local_deep_research.constants import ResearchStatus
        from local_deep_research.database.models import (
            QueuedResearch,
            ResearchHistory,
        )
        from local_deep_research.exceptions import SystemAtCapacityError
        from local_deep_research.web.queue.processor_v2 import (
            SPAWN_RETRY_LIMIT,
        )

        queued = Mock()
        queued.is_processing = False
        queued.research_id = "res-cap"

        research_row = Mock()

        def _make_query(model):
            q = Mock()
            q.filter_by.return_value = q
            q.order_by.return_value = q
            q.limit.return_value = q
            if model is QueuedResearch:
                q.all.return_value = [queued]
                q.first.return_value = queued
            elif model is ResearchHistory:
                q.first.return_value = research_row
            else:
                q.all.return_value = []
                q.first.return_value = None
            return q

        mock_session = Mock()
        mock_session.query.side_effect = _make_query

        mock_queue_service = Mock()
        processor._start_research = Mock(
            side_effect=SystemAtCapacityError("at capacity")
        )
        processor.notify_research_failed = Mock()

        # Run well past SPAWN_RETRY_LIMIT — capacity rejection must never
        # accumulate toward the limit no matter how many ticks pass.
        for _ in range(SPAWN_RETRY_LIMIT + 2):
            processor._start_queued_researches(
                mock_session, mock_queue_service, "alice", "pw", 1
            )

        # Never counted toward the retry limit, never marked FAILED, never
        # deleted, never notified — left QUEUED (is_processing reset) so the
        # next tick can retry once capacity frees up.
        assert "res-cap" not in processor._spawn_retry_counts
        processor.notify_research_failed.assert_not_called()
        mock_session.delete.assert_not_called()
        assert research_row.status != ResearchStatus.FAILED
        assert queued.is_processing is False

    def test_counter_resets_on_successful_start(self, processor):
        """If a research eventually succeeds after prior failures, the
        retry counter entry is popped so a future fresh failure gets a
        full retry budget."""
        queued = Mock()
        queued.is_processing = False
        queued.research_id = "res-4"

        mock_session = Mock()
        mock_query = Mock()
        mock_query.filter_by.return_value = mock_query
        mock_query.order_by.return_value = mock_query
        mock_query.limit.return_value = mock_query
        mock_query.all.return_value = [queued]
        mock_query.first.return_value = queued
        mock_session.query.return_value = mock_query

        mock_queue_service = Mock()
        processor._spawn_retry_counts["res-4"] = 2  # prior failures
        processor._start_research = Mock()  # success this time

        processor._start_queued_researches(
            mock_session, mock_queue_service, "alice", "pw", 1
        )

        assert "res-4" not in processor._spawn_retry_counts
        mock_session.delete.assert_called_with(queued)

    def test_success_deletes_queued_record(self, processor):
        """On success, the queued_research is deleted from DB."""
        queued = Mock()
        queued.is_processing = False
        queued.research_id = "res-3"

        mock_session = Mock()
        mock_query = Mock()
        mock_query.filter_by.return_value = mock_query
        mock_query.order_by.return_value = mock_query
        mock_query.limit.return_value = mock_query
        mock_query.all.return_value = [queued]
        mock_session.query.return_value = mock_query

        mock_queue_service = Mock()
        processor._start_research = Mock()  # success

        processor._start_queued_researches(
            mock_session, mock_queue_service, "alice", "pw", 1
        )

        mock_session.delete.assert_called_once_with(queued)

    def test_respects_available_slots_limit(self, processor):
        """The .limit() call should use available_slots."""
        mock_session = Mock()
        mock_query = Mock()
        mock_query.filter_by.return_value = mock_query
        mock_query.order_by.return_value = mock_query
        mock_query.limit.return_value = mock_query
        mock_query.all.return_value = []
        mock_session.query.return_value = mock_query

        mock_queue_service = Mock()

        processor._start_queued_researches(
            mock_session, mock_queue_service, "alice", "pw", 5
        )

        mock_query.limit.assert_called_with(5)

    def test_skip_when_already_claimed_by_another_worker(self, processor):
        """When the atomic UPDATE matches zero rows (another worker
        already flipped is_processing to True since our SELECT), we must
        skip the item without starting or deleting it."""
        queued = Mock()
        queued.is_processing = False
        queued.research_id = "res-claimed"

        mock_session = Mock()
        mock_query = Mock()
        mock_query.filter_by.return_value = mock_query
        mock_query.order_by.return_value = mock_query
        mock_query.limit.return_value = mock_query
        mock_query.all.return_value = [queued]
        # Simulate lost claim race: another worker beat us to it.
        mock_query.update.return_value = 0
        mock_session.query.return_value = mock_query

        mock_queue_service = Mock()
        processor._start_research = Mock()

        processor._start_queued_researches(
            mock_session, mock_queue_service, "alice", "pw", 1
        )

        processor._start_research.assert_not_called()
        mock_session.delete.assert_not_called()
        mock_session.refresh.assert_not_called()
        mock_queue_service.update_task_status.assert_not_called()

    def test_duplicate_research_error_deletes_queue_row_and_does_not_retry(
        self, processor
    ):
        """DuplicateResearchError means a live thread already exists (prior
        attempt's post-spawn commit failed). The queue row must be deleted,
        the retry counter cleared, and the failure-notification path must
        NOT run — that would terminate-status a research that is actually
        running."""
        from local_deep_research.exceptions import DuplicateResearchError

        queued = Mock()
        queued.is_processing = False
        queued.research_id = "res-dup"

        mock_session = Mock()
        mock_query = Mock()
        mock_query.filter_by.return_value = mock_query
        mock_query.order_by.return_value = mock_query
        mock_query.limit.return_value = mock_query
        mock_query.all.return_value = [queued]
        # The dup-cleanup path re-queries the queued row before deleting.
        mock_query.first.return_value = queued
        mock_session.query.return_value = mock_query

        mock_queue_service = Mock()
        processor._start_research = Mock(
            side_effect=DuplicateResearchError("thread already live")
        )
        processor.notify_research_failed = Mock()

        processor._start_queued_researches(
            mock_session, mock_queue_service, "alice", "pw", 1
        )

        # Queue row deleted, counter cleared, no failure notification.
        mock_session.delete.assert_called_with(queued)
        assert "res-dup" not in processor._spawn_retry_counts
        processor.notify_research_failed.assert_not_called()

    def test_duplicate_research_error_does_not_mark_failed_at_retry_limit(
        self, processor
    ):
        """Even when the retry counter is already at SPAWN_RETRY_LIMIT - 1
        (i.e. one more bump would trigger the terminal FAILED path),
        DuplicateResearchError must NOT increment the counter or mark
        ResearchHistory.status = FAILED. A live thread is running and
        will write its own terminal status."""
        from local_deep_research.database.models import (
            QueuedResearch,
            ResearchHistory,
        )
        from local_deep_research.exceptions import DuplicateResearchError
        from local_deep_research.web.queue.processor_v2 import (
            SPAWN_RETRY_LIMIT,
        )

        queued = Mock()
        queued.is_processing = False
        queued.research_id = "res-dup-limit"

        research_row = Mock()
        # Distinct status sentinel we can check was not overwritten.
        research_row.status = "NOT-FAILED-SENTINEL"

        def _make_query(model):
            q = Mock()
            q.filter_by.return_value = q
            q.order_by.return_value = q
            q.limit.return_value = q
            if model is QueuedResearch:
                q.all.return_value = [queued]
                q.first.return_value = queued
            elif model is ResearchHistory:
                q.first.return_value = research_row
            else:
                q.all.return_value = []
                q.first.return_value = None
            return q

        mock_session = Mock()
        mock_session.query.side_effect = _make_query

        mock_queue_service = Mock()
        processor._spawn_retry_counts["res-dup-limit"] = SPAWN_RETRY_LIMIT - 1
        processor._start_research = Mock(
            side_effect=DuplicateResearchError("thread already live")
        )
        processor.notify_research_failed = Mock()

        processor._start_queued_researches(
            mock_session, mock_queue_service, "alice", "pw", 1
        )

        # Counter cleared, not incremented to SPAWN_RETRY_LIMIT.
        assert "res-dup-limit" not in processor._spawn_retry_counts
        # ResearchHistory.status untouched — live thread owns it.
        assert research_row.status == "NOT-FAILED-SENTINEL"
        # No failure notification sent to the user.
        processor.notify_research_failed.assert_not_called()

    def test_transient_failure_then_duplicate_recovers_cleanly(self, processor):
        """End-to-end of the CRITICAL scenario this PR fixes: first tick's
        spawn succeeds but the post-spawn commit fails (simulated as a
        RuntimeError from _start_research); second tick finds the live
        thread and gets DuplicateResearchError. The dup branch must
        delete the queue row, clear the counter, and NOT mark FAILED —
        exactly what the original retry loop failed to do."""
        from local_deep_research.database.models import (
            QueuedResearch,
            ResearchHistory,
        )
        from local_deep_research.exceptions import DuplicateResearchError

        queued = Mock()
        queued.is_processing = False
        queued.research_id = "res-recover"

        research_row = Mock()
        # Sentinel so we can prove status was never overwritten to FAILED.
        research_row.status = "LIVE-THREAD-OWNED-SENTINEL"

        def _make_query(model):
            q = Mock()
            q.filter_by.return_value = q
            q.order_by.return_value = q
            q.limit.return_value = q
            if model is QueuedResearch:
                q.all.return_value = [queued]
                q.first.return_value = queued
            elif model is ResearchHistory:
                q.first.return_value = research_row
            else:
                q.all.return_value = []
                q.first.return_value = None
            return q

        mock_session = Mock()
        mock_session.query.side_effect = _make_query

        mock_queue_service = Mock()
        processor.notify_research_failed = Mock()

        # Tick 1: simulate post-spawn commit failure — _start_research
        # raised after the thread was already live. The generic except
        # path bumps the counter and leaves the row for retry.
        processor._start_research = Mock(
            side_effect=RuntimeError("post-spawn commit failed")
        )
        processor._start_queued_researches(
            mock_session, mock_queue_service, "alice", "pw", 1
        )
        assert processor._spawn_retry_counts["res-recover"] == 1
        processor.notify_research_failed.assert_not_called()

        # Tick 2: the live thread from tick 1 is still running, so
        # start_research_process raises DuplicateResearchError.
        processor._start_research = Mock(
            side_effect=DuplicateResearchError("thread already live")
        )
        processor._start_queued_researches(
            mock_session, mock_queue_service, "alice", "pw", 1
        )

        # Dup branch: counter cleared, queue row deleted, FAILED never
        # written, user never notified of a failure that didn't happen.
        assert "res-recover" not in processor._spawn_retry_counts
        assert research_row.status == "LIVE-THREAD-OWNED-SENTINEL"
        processor.notify_research_failed.assert_not_called()
        # Queue row was deleted (any call with the queued row counts).
        assert any(
            call.args and call.args[0] is queued
            for call in mock_session.delete.call_args_list
        )


# ---------------------------------------------------------------------------
# _reclaim_stranded_queue_rows — crash/restart recovery
# ---------------------------------------------------------------------------


class TestReclaimStrandedQueueRows:
    """Verify the reclaim pass recovers queue rows stranded by a crash
    between pre-spawn commit and queue-row deletion."""

    def _make_session_with_stranded_rows(self, stranded_rows, research_rows):
        """Build a mock session that returns ``stranded_rows`` for the
        QueuedResearch is_processing=True query and per-id lookups for
        ResearchHistory."""
        from local_deep_research.database.models import (
            QueuedResearch,
            ResearchHistory,
        )

        def _make_query(model):
            q = Mock()
            q.filter_by.return_value = q
            q.order_by.return_value = q
            q.limit.return_value = q
            if model is QueuedResearch:
                q.all.return_value = stranded_rows
                q.first.return_value = None
            elif model is ResearchHistory:

                def _first():
                    last_filter = q.filter_by.call_args
                    rid = last_filter.kwargs.get("id") if last_filter else None
                    return research_rows.get(rid)

                q.first.side_effect = _first
            else:
                q.all.return_value = []
                q.first.return_value = None
            return q

        session = Mock()
        session.query.side_effect = _make_query
        return session

    def test_reclaims_row_with_no_live_thread_and_in_progress_status(
        self, processor
    ):
        """After a crash, a row with is_processing=True + status=IN_PROGRESS
        + no live thread must be reverted to is_processing=False +
        status=QUEUED."""
        from local_deep_research.constants import ResearchStatus

        stranded = Mock()
        stranded.research_id = "res-stranded"
        stranded.is_processing = True

        research_row = Mock()
        research_row.status = ResearchStatus.IN_PROGRESS

        mock_session = self._make_session_with_stranded_rows(
            [stranded], {"res-stranded": research_row}
        )

        with patch(
            "local_deep_research.web.routes.globals.is_research_active",
            return_value=False,
        ):
            reclaimed = processor._reclaim_stranded_queue_rows(
                mock_session, "alice"
            )

        assert reclaimed == 1
        assert stranded.is_processing is False
        assert research_row.status == ResearchStatus.QUEUED
        assert mock_session.commit.called

    def test_does_not_reclaim_row_with_live_thread(self, processor):
        """A row whose research_id is in _active_research (live thread)
        must NOT be reclaimed — it's legitimately in-flight."""
        from local_deep_research.constants import ResearchStatus

        stranded = Mock()
        stranded.research_id = "res-live"
        stranded.is_processing = True

        research_row = Mock()
        research_row.status = ResearchStatus.IN_PROGRESS

        mock_session = self._make_session_with_stranded_rows(
            [stranded], {"res-live": research_row}
        )

        with patch(
            "local_deep_research.web.routes.globals.is_research_active",
            return_value=True,
        ):
            reclaimed = processor._reclaim_stranded_queue_rows(
                mock_session, "alice"
            )

        assert reclaimed == 0
        assert stranded.is_processing is True
        assert research_row.status == ResearchStatus.IN_PROGRESS
        # No commit needed when nothing was reclaimed.
        assert not mock_session.commit.called

    def test_reclaims_row_without_touching_terminal_status(self, processor):
        """If a stranded row's ResearchHistory is already COMPLETED/FAILED
        (thread finished cleanly before the crash), only reset
        is_processing; do NOT overwrite the terminal status."""
        from local_deep_research.constants import ResearchStatus

        stranded = Mock()
        stranded.research_id = "res-done"
        stranded.is_processing = True

        research_row = Mock()
        research_row.status = ResearchStatus.COMPLETED

        mock_session = self._make_session_with_stranded_rows(
            [stranded], {"res-done": research_row}
        )

        with patch(
            "local_deep_research.web.routes.globals.is_research_active",
            return_value=False,
        ):
            reclaimed = processor._reclaim_stranded_queue_rows(
                mock_session, "alice"
            )

        assert reclaimed == 1
        assert stranded.is_processing is False
        # Terminal status untouched.
        assert research_row.status == ResearchStatus.COMPLETED

    def test_no_stranded_rows_is_cheap_noop(self, processor):
        """With zero stranded rows, no commit should fire."""
        mock_session = self._make_session_with_stranded_rows([], {})

        with patch(
            "local_deep_research.web.routes.globals.is_research_active",
            return_value=False,
        ):
            reclaimed = processor._reclaim_stranded_queue_rows(
                mock_session, "alice"
            )

        assert reclaimed == 0
        assert not mock_session.commit.called

    def test_start_queued_researches_calls_reclaim_before_select(
        self, processor
    ):
        """_start_queued_researches must invoke _reclaim_stranded_queue_rows
        BEFORE the main QueuedResearch SELECT. If the order were swapped,
        rows just reclaimed (is_processing flipped to False) would still
        be invisible to the SELECT in the same tick — which is the entire
        point of the feature. Use a shared ``call_order`` list to assert
        the ordering, not just that both calls happened."""
        call_order = []

        mock_session = Mock()
        mock_query = Mock()
        mock_query.filter_by.return_value = mock_query
        mock_query.order_by.return_value = mock_query
        mock_query.limit.return_value = mock_query
        mock_query.all.return_value = []

        def _query_side_effect(*args, **kwargs):
            call_order.append("query")
            return mock_query

        mock_session.query.side_effect = _query_side_effect

        def _reclaim_side_effect(*args, **kwargs):
            call_order.append("reclaim")
            return 0

        mock_queue_service = Mock()
        processor._reclaim_stranded_queue_rows = Mock(
            side_effect=_reclaim_side_effect
        )

        processor._start_queued_researches(
            mock_session, mock_queue_service, "alice", "pw", 1
        )

        processor._reclaim_stranded_queue_rows.assert_called_once_with(
            mock_session, "alice"
        )
        # Ordering invariant: reclaim must land before the first query.
        assert call_order[0] == "reclaim", (
            f"Expected 'reclaim' first, got order: {call_order}"
        )


# ---------------------------------------------------------------------------
# _start_research_directly cleanup
# ---------------------------------------------------------------------------


class TestStartResearchDirectlyCleanup:
    """When start_research_process throws, verify UserActiveResearch deleted."""

    @patch("local_deep_research.web.queue.processor_v2.UserQueueService")
    @patch("local_deep_research.web.queue.processor_v2.get_user_db_session")
    @patch("local_deep_research.web.queue.processor_v2.start_research_process")
    def test_cleanup_on_start_failure(
        self, mock_start, mock_get_session, mock_qs, processor
    ):
        """When start_research_process raises, the active record must be
        deleted AND the ResearchHistory row marked FAILED."""
        from local_deep_research.constants import ResearchStatus
        from local_deep_research.database.models import (
            ResearchHistory,
            UserActiveResearch,
        )

        active_record = Mock()
        research_row = Mock()

        session1 = Mock()  # Create record session
        session2 = Mock()  # Cleanup session

        def _query(model):
            q = Mock()
            q.filter_by.return_value = q
            q.filter.return_value = q
            if model is UserActiveResearch:
                q.first.return_value = active_record
            elif model is ResearchHistory:
                q.first.return_value = research_row
            else:
                q.first.return_value = None
            return q

        session2.query.side_effect = _query

        sessions = [session1, session2]
        call_count = [0]

        @contextmanager
        def fake_session(*args, **kwargs):
            idx = min(call_count[0], len(sessions) - 1)
            call_count[0] += 1
            yield sessions[idx]

        mock_get_session.side_effect = fake_session
        mock_start.side_effect = RuntimeError("process failed")

        processor._start_research_directly(
            "alice",
            "res-1",
            "password",
            query="test",
            mode="quick",
        )

        # Cleanup session deletes the active record AND marks FAILED.
        session2.delete.assert_called_once_with(active_record)
        assert research_row.status == ResearchStatus.FAILED
        session2.commit.assert_called()

    @patch("local_deep_research.web.queue.processor_v2.get_user_db_session")
    @patch("local_deep_research.web.queue.processor_v2.start_research_process")
    def test_success_updates_thread_id(
        self, mock_start, mock_get_session, processor
    ):
        """On success, thread ID is updated in active record."""
        mock_thread = Mock()
        mock_thread.ident = 12345
        mock_start.return_value = mock_thread

        active_record = Mock()
        mock_session = Mock()
        mock_query = Mock()
        mock_query.filter_by.return_value = mock_query
        mock_query.first.return_value = active_record
        mock_session.query.return_value = mock_query

        @contextmanager
        def fake_session(*args, **kwargs):
            yield mock_session

        mock_get_session.side_effect = fake_session

        processor._start_research_directly(
            "alice",
            "res-1",
            "password",
            query="test",
            mode="quick",
        )

        assert active_record.thread_id == "12345"


# ---------------------------------------------------------------------------
# process_user_request
# ---------------------------------------------------------------------------


class TestProcessUserRequest:
    """Tests for process_user_request return values."""

    @patch("local_deep_research.web.queue.processor_v2.db_manager")
    @patch("local_deep_research.web.queue.processor_v2.session_password_store")
    @patch("local_deep_research.web.queue.processor_v2.get_user_db_session")
    def test_no_password_returns_zero(
        self, mock_session, mock_pw_store, mock_db, processor
    ):
        """When no password available, returns 0 (line 686)."""
        mock_pw_store.get_session_password.return_value = None

        result = processor.process_user_request("alice", "sess-1")
        assert result == 0

    @patch("local_deep_research.web.queue.processor_v2.db_manager")
    @patch("local_deep_research.web.queue.processor_v2.session_password_store")
    @patch("local_deep_research.web.queue.processor_v2.get_user_db_session")
    def test_exception_returns_zero(
        self, mock_session, mock_pw_store, mock_db, processor
    ):
        """When exception occurs, returns 0 (line 690)."""
        mock_pw_store.get_session_password.return_value = "pw"
        mock_db.open_user_database.side_effect = RuntimeError("db error")

        result = processor.process_user_request("alice", "sess-1")
        assert result == 0

    @patch("local_deep_research.web.queue.processor_v2.db_manager")
    @patch("local_deep_research.web.queue.processor_v2.session_password_store")
    @patch("local_deep_research.web.queue.processor_v2.get_user_db_session")
    def test_returns_queued_count(
        self, mock_get_session, mock_pw_store, mock_db, processor
    ):
        """When queue has items, returns queued count (line 684)."""
        mock_pw_store.get_session_password.return_value = "pw"
        mock_db.open_user_database.return_value = Mock()

        mock_queue_service = Mock()
        mock_queue_service.get_queue_status.return_value = {"queued_tasks": 3}

        mock_session = Mock()

        @contextmanager
        def fake_session(*args, **kwargs):
            yield mock_session

        mock_get_session.side_effect = fake_session

        with patch(
            "local_deep_research.web.queue.processor_v2.UserQueueService",
            return_value=mock_queue_service,
        ):
            result = processor.process_user_request("alice", "sess-1")

        assert result == 3

    @patch("local_deep_research.web.queue.processor_v2.db_manager")
    @patch("local_deep_research.web.queue.processor_v2.session_password_store")
    @patch("local_deep_research.web.queue.processor_v2.get_user_db_session")
    def test_empty_queue_returns_zero(
        self, mock_get_session, mock_pw_store, mock_db, processor
    ):
        """When queue is empty, returns 0."""
        mock_pw_store.get_session_password.return_value = "pw"
        mock_db.open_user_database.return_value = Mock()

        mock_queue_service = Mock()
        mock_queue_service.get_queue_status.return_value = {"queued_tasks": 0}

        mock_session = Mock()

        @contextmanager
        def fake_session(*args, **kwargs):
            yield mock_session

        mock_get_session.side_effect = fake_session

        with patch(
            "local_deep_research.web.queue.processor_v2.UserQueueService",
            return_value=mock_queue_service,
        ):
            result = processor.process_user_request("alice", "sess-1")

        assert result == 0


# ---------------------------------------------------------------------------
# notify_user_activity concurrency
# ---------------------------------------------------------------------------


class TestNotifyUserActivityConcurrency:
    """Concurrent notify_user_activity calls from 20 threads."""

    def test_concurrent_notify_all_users_registered(self, processor):
        """All users registered via concurrent notify_user_activity calls."""
        barrier = threading.Barrier(20)
        errors = []

        def worker(i):
            try:
                barrier.wait(timeout=5)
                processor.notify_user_activity(f"user-{i}", f"sess-{i}")
            except Exception as e:
                errors.append(e)

        threads = [
            threading.Thread(target=worker, args=(i,)) for i in range(20)
        ]
        for t in threads:
            t.start()
        for t in threads:
            t.join(timeout=10)

        assert not errors
        assert len(processor._users_to_check) == 20
        for i in range(20):
            assert (f"user-{i}", f"sess-{i}") in processor._users_to_check

    def test_duplicate_user_session_deduplicated(self, processor):
        """Same user:session added multiple times is stored once (set behavior)."""
        for _ in range(5):
            processor.notify_user_activity("alice", "sess-1")

        assert len(processor._users_to_check) == 1
        assert ("alice", "sess-1") in processor._users_to_check


# ---------------------------------------------------------------------------
# _spawn_retry_counts concurrent access
# ---------------------------------------------------------------------------


class TestSpawnRetryCountsConcurrency:
    """Verify _spawn_retry_counts increments are atomic under contention."""

    def test_concurrent_increments_do_not_lose_updates(self, processor):
        """Calls the production ``_bump_spawn_retry_count`` helper from
        many threads. Without the lock inside that helper, the
        read-modify-write loses increments under contention and the
        final count is less than ``n_threads``. With the lock, the final
        count must equal ``n_threads``. This is a real mutation test of
        the production lock — removing ``with self._spawn_retry_counts_lock:``
        from the helper makes this test fail."""
        n_threads = 50
        research_id = "contended-id"
        barrier = threading.Barrier(n_threads)
        returned_values = []
        returned_lock = threading.Lock()

        def worker():
            barrier.wait(timeout=5)
            # Call the production increment path, not the lock directly.
            attempts = processor._bump_spawn_retry_count(research_id)
            with returned_lock:
                returned_values.append(attempts)

        threads = [threading.Thread(target=worker) for _ in range(n_threads)]
        for t in threads:
            t.start()
        for t in threads:
            t.join(timeout=10)

        # Final stored count must equal the number of incrementers.
        assert processor._spawn_retry_counts[research_id] == n_threads
        # Returned values must be 1..n_threads with no duplicates — any
        # duplicate would mean two callers observed the same prior count,
        # i.e. a lost update.
        assert sorted(returned_values) == list(range(1, n_threads + 1))


# ---------------------------------------------------------------------------
# Start/stop lifecycle
# ---------------------------------------------------------------------------


class TestProcessorLifecycle:
    """Basic start/stop behavior."""

    def test_start_sets_running(self, processor):
        processor.start()
        assert processor.running is True
        processor.stop()
        assert processor.running is False

    def test_double_start_is_noop(self, processor):
        processor.start()
        processor.start()  # should not raise
        processor.stop()

    def test_stop_without_start(self, processor):
        """Stopping without starting should not raise."""
        processor.stop()
        assert processor.running is False