learningcircuit--local-deep-research/tests/library/test_auto_indexing.py

"""Tests for automatic RAG indexing functionality."""

import threading
import time
from unittest.mock import MagicMock, patch

import pytest


@pytest.fixture(autouse=True)
def _reset_pending_auto_index_jobs():
    """Save/restore the module-global pending-jobs counter around each test.

    Several tests mock ``_get_auto_index_executor`` with a no-op ``submit``,
    which reserves a slot (``_pending_auto_index_jobs += 1``) that the (never
    run) worker never releases. Without this fixture the counter leaks +1 per
    such test and the saturation/concurrency tests become order-dependent.
    This restores the counter to its pre-test value after every test, keeping
    the suite order-independent.
    """
    import local_deep_research.research_library.routes.rag_routes as rag_module

    with rag_module._pending_auto_index_lock:
        saved = rag_module._pending_auto_index_jobs
    try:
        yield
    finally:
        with rag_module._pending_auto_index_lock:
            rag_module._pending_auto_index_jobs = saved


class TestAutoIndexingSetting:
    """Test the auto-indexing setting."""

    def test_auto_index_setting_exists_in_defaults(self):
        """Test that the auto_index_enabled setting is defined in defaults."""
        import json
        from pathlib import Path

        defaults_path = Path(
            "src/local_deep_research/defaults/default_settings.json"
        )
        with open(defaults_path) as f:
            defaults = json.load(f)

        assert "research_library.auto_index_enabled" in defaults
        setting = defaults["research_library.auto_index_enabled"]
        assert setting["ui_element"] == "checkbox"
        assert setting["value"] is False  # Default is disabled (opt-in)
        assert setting["category"] == "research_library"


class TestTriggerAutoIndex:
    """Test the trigger_auto_index function."""

    def test_trigger_auto_index_skips_when_disabled(self):
        """Test that auto-indexing is skipped when disabled in settings."""
        from local_deep_research.research_library.routes.rag_routes import (
            trigger_auto_index,
        )

        with patch(
            "local_deep_research.database.session_context.get_user_db_session"
        ) as mock_session:
            mock_db = MagicMock()
            mock_session.return_value.__enter__.return_value = mock_db

            mock_settings = MagicMock()
            mock_settings.get_bool_setting.return_value = False

            with patch(
                "local_deep_research.research_library.routes.rag_routes.SettingsManager",
                return_value=mock_settings,
            ):
                with patch(
                    "local_deep_research.research_library.routes.rag_routes._get_auto_index_executor"
                ) as mock_get_executor:
                    trigger_auto_index(
                        document_ids=["doc1"],
                        collection_id="coll1",
                        username="testuser",
                        db_password="testpass",
                    )

                    # Executor should NOT be called when disabled
                    mock_get_executor.assert_not_called()

    def test_trigger_auto_index_submits_to_executor_when_enabled(self):
        """Test that auto-indexing submits to executor when enabled."""
        from local_deep_research.research_library.routes.rag_routes import (
            trigger_auto_index,
        )

        with patch(
            "local_deep_research.database.session_context.get_user_db_session"
        ) as mock_session:
            mock_db = MagicMock()
            mock_session.return_value.__enter__.return_value = mock_db

            mock_settings = MagicMock()
            mock_settings.get_bool_setting.return_value = True

            with patch(
                "local_deep_research.research_library.routes.rag_routes.SettingsManager",
                return_value=mock_settings,
            ):
                with patch(
                    "local_deep_research.research_library.routes.rag_routes._get_auto_index_executor"
                ) as mock_get_executor:
                    mock_executor = MagicMock()
                    mock_get_executor.return_value = mock_executor

                    trigger_auto_index(
                        document_ids=["doc1", "doc2"],
                        collection_id="coll1",
                        username="testuser",
                        db_password="testpass",
                    )

                    # Executor should be obtained and submit called
                    mock_get_executor.assert_called_once()
                    mock_executor.submit.assert_called_once()

    def test_trigger_auto_index_skips_empty_document_list(self):
        """Test that auto-indexing is skipped when no documents provided."""
        from local_deep_research.research_library.routes.rag_routes import (
            trigger_auto_index,
        )

        with patch(
            "local_deep_research.database.session_context.get_user_db_session"
        ) as mock_session:
            # Should not even try to access the database
            trigger_auto_index(
                document_ids=[],
                collection_id="coll1",
                username="testuser",
                db_password="testpass",
            )

            mock_session.assert_not_called()

    def test_trigger_auto_index_skips_on_settings_exception(self):
        """Test that auto-indexing is skipped when settings check raises exception."""
        from local_deep_research.research_library.routes.rag_routes import (
            trigger_auto_index,
        )

        with patch(
            "local_deep_research.database.session_context.get_user_db_session"
        ) as mock_session:
            # Simulate database error
            mock_session.side_effect = Exception("Database connection failed")

            with patch(
                "local_deep_research.research_library.routes.rag_routes._get_auto_index_executor"
            ) as mock_get_executor:
                # Should not raise, just skip
                trigger_auto_index(
                    document_ids=["doc1"],
                    collection_id="coll1",
                    username="testuser",
                    db_password="testpass",
                )

                # Executor should NOT be called when settings check fails
                mock_get_executor.assert_not_called()

    def test_trigger_auto_index_passes_correct_arguments(self):
        """Backpressure wrapper still delegates correct args to the worker.

        The backpressure change submits a ``_wrapped_worker`` closure rather
        than ``_auto_index_documents_worker`` directly. This test verifies the
        full contract of that wrapper:

        * a queue slot is reserved before the job is submitted,
        * the submitted callable still forwards the exact arguments to the
          underlying ``_auto_index_documents_worker``, and
        * the reserved slot is released once the wrapped worker finishes.
        """
        import local_deep_research.research_library.routes.rag_routes as rag_module
        from local_deep_research.research_library.routes.rag_routes import (
            trigger_auto_index,
        )

        with patch(
            "local_deep_research.database.session_context.get_user_db_session"
        ) as mock_session:
            mock_db = MagicMock()
            mock_session.return_value.__enter__.return_value = mock_db

            mock_settings = MagicMock()
            mock_settings.get_bool_setting.return_value = True

            with patch(
                "local_deep_research.research_library.routes.rag_routes.SettingsManager",
                return_value=mock_settings,
            ):
                with patch(
                    "local_deep_research.research_library.routes.rag_routes._get_auto_index_executor"
                ) as mock_get_executor:
                    mock_executor = MagicMock()
                    mock_get_executor.return_value = mock_executor

                    with patch(
                        "local_deep_research.research_library.routes.rag_routes._auto_index_documents_worker"
                    ) as mock_worker:
                        pending_before = rag_module._pending_auto_index_jobs

                        trigger_auto_index(
                            document_ids=["doc1", "doc2", "doc3"],
                            collection_id="my_collection",
                            username="alice",
                            db_password="secret123",
                        )

                        # A queue slot must be reserved before submission, so
                        # the pending counter is incremented while the (not yet
                        # run) wrapped worker holds the slot.
                        assert (
                            rag_module._pending_auto_index_jobs
                            == pending_before + 1
                        )

                        # A wrapper closure is submitted, NOT the worker
                        # directly (that is the whole point of the wrapper).
                        mock_executor.submit.assert_called_once()
                        call_args = mock_executor.submit.call_args
                        submitted_callable = call_args[0][0]
                        assert submitted_callable is not mock_worker
                        assert callable(submitted_callable)

                        # The wrapper forwards the original arguments unchanged.
                        assert call_args[0][1] == ["doc1", "doc2", "doc3"]
                        assert call_args[0][2] == "my_collection"
                        assert call_args[0][3] == "alice"
                        assert call_args[0][4] == "secret123"

                        # The underlying worker has not run yet: the executor
                        # is mocked, so submit did not invoke the callable.
                        mock_worker.assert_not_called()

                        # Simulate the executor running the submitted job. The
                        # wrapper must forward the args to the real worker and
                        # then release the reserved slot.
                        submitted_callable(*call_args[0][1:])

                        mock_worker.assert_called_once_with(
                            ["doc1", "doc2", "doc3"],
                            "my_collection",
                            "alice",
                            "secret123",
                        )

                        # Slot released once the wrapped worker completed.
                        assert (
                            rag_module._pending_auto_index_jobs
                            == pending_before
                        )

    def test_trigger_auto_index_with_single_document(self):
        """Test that auto-indexing works with a single document."""
        from local_deep_research.research_library.routes.rag_routes import (
            trigger_auto_index,
        )

        with patch(
            "local_deep_research.database.session_context.get_user_db_session"
        ) as mock_session:
            mock_db = MagicMock()
            mock_session.return_value.__enter__.return_value = mock_db

            mock_settings = MagicMock()
            mock_settings.get_bool_setting.return_value = True

            with patch(
                "local_deep_research.research_library.routes.rag_routes.SettingsManager",
                return_value=mock_settings,
            ):
                with patch(
                    "local_deep_research.research_library.routes.rag_routes._get_auto_index_executor"
                ) as mock_get_executor:
                    mock_executor = MagicMock()
                    mock_get_executor.return_value = mock_executor

                    trigger_auto_index(
                        document_ids=["single_doc"],
                        collection_id="coll1",
                        username="testuser",
                        db_password="testpass",
                    )

                    mock_executor.submit.assert_called_once()

    def test_trigger_auto_index_with_many_documents(self):
        """Test that auto-indexing works with many documents."""
        from local_deep_research.research_library.routes.rag_routes import (
            trigger_auto_index,
        )

        with patch(
            "local_deep_research.database.session_context.get_user_db_session"
        ) as mock_session:
            mock_db = MagicMock()
            mock_session.return_value.__enter__.return_value = mock_db

            mock_settings = MagicMock()
            mock_settings.get_bool_setting.return_value = True

            with patch(
                "local_deep_research.research_library.routes.rag_routes.SettingsManager",
                return_value=mock_settings,
            ):
                with patch(
                    "local_deep_research.research_library.routes.rag_routes._get_auto_index_executor"
                ) as mock_get_executor:
                    mock_executor = MagicMock()
                    mock_get_executor.return_value = mock_executor

                    # Submit 100 documents
                    doc_ids = [f"doc_{i}" for i in range(100)]
                    trigger_auto_index(
                        document_ids=doc_ids,
                        collection_id="coll1",
                        username="testuser",
                        db_password="testpass",
                    )

                    mock_executor.submit.assert_called_once()
                    call_args = mock_executor.submit.call_args
                    assert len(call_args[0][1]) == 100

    def test_trigger_auto_index_drops_job_when_queue_saturated(self):
        """Saturated queue: the job is dropped, not submitted, and warned.

        When ``_pending_auto_index_jobs`` is already at the bound, the
        backpressure guard must refuse the new submission entirely: it must
        NOT touch the executor (no ``submit``), it must emit a warning, and it
        must leave the saturation counter untouched (no slot consumed, so the
        upload path that called it is unaffected and still succeeded).
        """
        import local_deep_research.research_library.routes.rag_routes as rag_module
        from local_deep_research.research_library.routes.rag_routes import (
            trigger_auto_index,
        )

        with patch(
            "local_deep_research.database.session_context.get_user_db_session"
        ) as mock_session:
            mock_db = MagicMock()
            mock_session.return_value.__enter__.return_value = mock_db

            mock_settings = MagicMock()
            mock_settings.get_bool_setting.return_value = True

            saturated = rag_module._MAX_PENDING_AUTO_INDEX_JOBS
            original_pending = rag_module._pending_auto_index_jobs
            rag_module._pending_auto_index_jobs = saturated
            try:
                with patch(
                    "local_deep_research.research_library.routes.rag_routes.SettingsManager",
                    return_value=mock_settings,
                ):
                    with patch(
                        "local_deep_research.research_library.routes.rag_routes._get_auto_index_executor"
                    ) as mock_get_executor:
                        mock_executor = MagicMock()
                        mock_get_executor.return_value = mock_executor

                        with patch.object(
                            rag_module.logger, "warning"
                        ) as mock_warning:
                            # Must not raise; the upload already succeeded.
                            trigger_auto_index(
                                document_ids=["doc1", "doc2"],
                                collection_id="coll1",
                                username="testuser",
                                db_password="testpass",
                            )

                        # Job dropped: executor never touched, nothing queued.
                        mock_get_executor.assert_not_called()
                        mock_executor.submit.assert_not_called()

                        # A saturation warning was emitted, and it carries the
                        # diagnostic context an operator needs: the cap that was
                        # hit, the number of documents dropped, and which
                        # collection they belonged to.
                        mock_warning.assert_called_once()
                        warning_args = mock_warning.call_args.args
                        # The cap, the document count (2 docs passed), and the
                        # collection id must all be present in the positional
                        # args passed to logger.warning.
                        assert (
                            rag_module._MAX_PENDING_AUTO_INDEX_JOBS
                            in warning_args
                        )
                        assert 2 in warning_args  # len(["doc1", "doc2"])
                        assert "coll1" in warning_args

                        # Counter unchanged: a dropped job consumes no slot.
                        assert rag_module._pending_auto_index_jobs == saturated
            finally:
                rag_module._pending_auto_index_jobs = original_pending

    def test_trigger_auto_index_releases_slot_when_worker_raises(self):
        """Leak prevention: a slot is released even if the worker raises.

        The wrapped worker submitted to the executor wraps the real worker in
        ``try/finally`` so that the reserved slot is always released. If this
        did not happen, a worker that raised would permanently leak its slot
        and eventually saturate the queue forever. This asserts the counter
        returns to its baseline after the worker raises.
        """
        import local_deep_research.research_library.routes.rag_routes as rag_module
        from local_deep_research.research_library.routes.rag_routes import (
            trigger_auto_index,
        )

        with patch(
            "local_deep_research.database.session_context.get_user_db_session"
        ) as mock_session:
            mock_db = MagicMock()
            mock_session.return_value.__enter__.return_value = mock_db

            mock_settings = MagicMock()
            mock_settings.get_bool_setting.return_value = True

            with patch(
                "local_deep_research.research_library.routes.rag_routes.SettingsManager",
                return_value=mock_settings,
            ):
                with patch(
                    "local_deep_research.research_library.routes.rag_routes._get_auto_index_executor"
                ) as mock_get_executor:
                    mock_executor = MagicMock()
                    mock_get_executor.return_value = mock_executor

                    with patch(
                        "local_deep_research.research_library.routes.rag_routes._auto_index_documents_worker",
                        side_effect=RuntimeError("worker boom"),
                    ):
                        pending_before = rag_module._pending_auto_index_jobs

                        trigger_auto_index(
                            document_ids=["doc1"],
                            collection_id="coll1",
                            username="testuser",
                            db_password="testpass",
                        )

                        # Slot reserved while the (not yet run) wrapper holds it.
                        assert (
                            rag_module._pending_auto_index_jobs
                            == pending_before + 1
                        )

                        # Run the submitted wrapper; the underlying worker
                        # raises. The wrapper's finally must still release the
                        # slot, and the raise must not escape the wrapper in a
                        # way that prevents that release.
                        submitted_callable = mock_executor.submit.call_args[0][
                            0
                        ]
                        submitted_args = mock_executor.submit.call_args[0][1:]
                        try:
                            submitted_callable(*submitted_args)
                        except RuntimeError:
                            # The executor would capture this in the Future;
                            # what matters is the slot was released first.
                            pass

                        # Slot released back to baseline despite the failure.
                        assert (
                            rag_module._pending_auto_index_jobs
                            == pending_before
                        )

    def test_trigger_auto_index_releases_slot_and_swallows_submit_failure(
        self,
    ):
        """Submit failure must not turn a committed upload into a 500.

        If ``executor.submit`` itself fails (executor shut down, OOM, etc.)
        the wrapped worker never runs, so ``trigger_auto_index`` must release
        the reserved slot AND must NOT propagate the exception. Propagating
        would bubble up to the upload handler and return a 500 even though the
        documents were already committed, prompting the client to retry and
        create duplicates.
        """
        import local_deep_research.research_library.routes.rag_routes as rag_module
        from local_deep_research.research_library.routes.rag_routes import (
            trigger_auto_index,
        )

        with patch(
            "local_deep_research.database.session_context.get_user_db_session"
        ) as mock_session:
            mock_db = MagicMock()
            mock_session.return_value.__enter__.return_value = mock_db

            mock_settings = MagicMock()
            mock_settings.get_bool_setting.return_value = True

            with patch(
                "local_deep_research.research_library.routes.rag_routes.SettingsManager",
                return_value=mock_settings,
            ):
                with patch(
                    "local_deep_research.research_library.routes.rag_routes._get_auto_index_executor"
                ) as mock_get_executor:
                    mock_executor = MagicMock()
                    mock_executor.submit.side_effect = RuntimeError(
                        "cannot schedule new futures after shutdown"
                    )
                    mock_get_executor.return_value = mock_executor

                    pending_before = rag_module._pending_auto_index_jobs

                    # Must return normally: the exception is swallowed so the
                    # already-committed upload is not turned into a failure.
                    trigger_auto_index(
                        document_ids=["doc1"],
                        collection_id="coll1",
                        username="testuser",
                        db_password="testpass",
                    )

                    # submit was attempted and failed.
                    mock_executor.submit.assert_called_once()

                    # Reserved slot was released despite the submit failure,
                    # so a failed submit does not leak a slot.
                    assert rag_module._pending_auto_index_jobs == pending_before

    def test_trigger_auto_index_releases_slot_when_executor_build_fails(self):
        """Slot must be released if _get_auto_index_executor() itself raises.

        Building the executor (OS thread/mutex exhaustion) happens after the
        slot is reserved. It must be inside the same try/except as submit, so
        a failure there releases the slot and is swallowed — otherwise the
        reserved slot leaks permanently and erodes the queue bound.
        """
        import local_deep_research.research_library.routes.rag_routes as rag_module
        from local_deep_research.research_library.routes.rag_routes import (
            trigger_auto_index,
        )

        with patch(
            "local_deep_research.database.session_context.get_user_db_session"
        ) as mock_session:
            mock_db = MagicMock()
            mock_session.return_value.__enter__.return_value = mock_db

            mock_settings = MagicMock()
            mock_settings.get_bool_setting.return_value = True

            with patch(
                "local_deep_research.research_library.routes.rag_routes.SettingsManager",
                return_value=mock_settings,
            ):
                with patch(
                    "local_deep_research.research_library.routes.rag_routes._get_auto_index_executor",
                    side_effect=RuntimeError("cannot allocate thread"),
                ):
                    pending_before = rag_module._pending_auto_index_jobs

                    # Must return normally: the exception is swallowed.
                    trigger_auto_index(
                        document_ids=["doc1"],
                        collection_id="coll1",
                        username="testuser",
                        db_password="testpass",
                    )

                    # Reserved slot was released despite the executor-build
                    # failure, so it does not leak a slot.
                    assert rag_module._pending_auto_index_jobs == pending_before

    def test_trigger_auto_index_releases_slot_once_on_thread_start_race(self):
        """Thread-start race must release the reserved slot EXACTLY once.

        CPython's ThreadPoolExecutor.submit() enqueues the work item BEFORE it
        tries to spin up a worker thread. If thread-start raises
        RuntimeError("can't start new thread"), the item is already queued, so
        a live worker can run the wrapped worker (which releases the slot in
        its finally) AND the except block around submit() also fires. Without
        an idempotent release that is a DOUBLE release of one reservation,
        under-counting in-flight jobs and eroding the OOM bound.

        We reproduce that race by making submit() RUN the passed worker (so its
        finally releases) and THEN raise the thread-start RuntimeError (so the
        except path also runs). The slot must be released exactly once.

        Critically, we pre-set the baseline to a NON-ZERO value: at baseline 0
        the ``max(0, ...)`` floor in ``_release_auto_index_slot`` would clamp a
        double-release back to 0 and silently mask the bug. A non-zero baseline
        makes the second (erroneous) release observable as baseline - 1.
        """
        import local_deep_research.research_library.routes.rag_routes as rag_module
        from local_deep_research.research_library.routes.rag_routes import (
            trigger_auto_index,
        )

        with patch(
            "local_deep_research.database.session_context.get_user_db_session"
        ) as mock_session:
            mock_db = MagicMock()
            mock_session.return_value.__enter__.return_value = mock_db

            mock_settings = MagicMock()
            mock_settings.get_bool_setting.return_value = True

            # Non-zero baseline so a double-release is observable (the max(0,..)
            # floor would otherwise mask it at baseline 0).
            baseline = 5
            with rag_module._pending_auto_index_lock:
                rag_module._pending_auto_index_jobs = baseline

            def submit_runs_then_raises(fn, *args, **kwargs):
                # Mirror CPython: the work item is enqueued and a worker runs
                # it (releasing the slot in the wrapped worker's finally)...
                fn(*args, **kwargs)
                # ...and THEN thread-start fails, so the except path fires too.
                raise RuntimeError("can't start new thread")

            with patch(
                "local_deep_research.research_library.routes.rag_routes.SettingsManager",
                return_value=mock_settings,
            ):
                with patch(
                    "local_deep_research.research_library.routes.rag_routes._get_auto_index_executor"
                ) as mock_get_executor:
                    mock_executor = MagicMock()
                    mock_executor.submit.side_effect = submit_runs_then_raises
                    mock_get_executor.return_value = mock_executor

                    with patch(
                        "local_deep_research.research_library.routes.rag_routes._auto_index_documents_worker"
                    ):
                        # Must not raise: the submit failure is swallowed.
                        trigger_auto_index(
                            document_ids=["doc1"],
                            collection_id="coll1",
                            username="testuser",
                            db_password="testpass",
                        )

                    # The slot was released EXACTLY ONCE despite both the
                    # wrapped worker's finally and the except block firing, so
                    # the counter returns to exactly the baseline. If the
                    # release were not idempotent, it would be baseline - 1.
                    assert rag_module._pending_auto_index_jobs == baseline


class TestAutoIndexExecutor:
    """Test the ThreadPoolExecutor infrastructure for auto-indexing."""

    def test_get_auto_index_executor_returns_executor(self):
        """Test that _get_auto_index_executor returns a ThreadPoolExecutor."""
        from concurrent.futures import ThreadPoolExecutor

        from local_deep_research.research_library.routes.rag_routes import (
            _get_auto_index_executor,
        )

        executor = _get_auto_index_executor()
        assert isinstance(executor, ThreadPoolExecutor)

    def test_get_auto_index_executor_returns_same_instance(self):
        """Test that _get_auto_index_executor returns the same singleton instance."""
        from local_deep_research.research_library.routes.rag_routes import (
            _get_auto_index_executor,
        )

        executor1 = _get_auto_index_executor()
        executor2 = _get_auto_index_executor()
        assert executor1 is executor2

    def test_shutdown_auto_index_executor(self):
        """Test that _shutdown_auto_index_executor properly shuts down the executor."""
        import local_deep_research.research_library.routes.rag_routes as rag_module
        from local_deep_research.research_library.routes.rag_routes import (
            _get_auto_index_executor,
            _shutdown_auto_index_executor,
        )

        # Ensure executor exists
        executor = _get_auto_index_executor()
        assert executor is not None

        # Shutdown
        _shutdown_auto_index_executor()

        # Global should be None after shutdown
        assert rag_module._auto_index_executor is None

        # Getting executor again should create a new one
        new_executor = _get_auto_index_executor()
        assert new_executor is not None
        assert new_executor is not executor

    def test_shutdown_resets_pending_jobs_counter(self):
        """Shutdown must reset the pending-jobs counter.

        Otherwise a re-created executor (tests / WSGI reload) inherits a
        stale, possibly-saturated count and could refuse all future
        submissions.
        """
        import local_deep_research.research_library.routes.rag_routes as rag_module
        from local_deep_research.research_library.routes.rag_routes import (
            _shutdown_auto_index_executor,
        )

        # Simulate slots left reserved (e.g. an interrupted shutdown).
        with rag_module._pending_auto_index_lock:
            rag_module._pending_auto_index_jobs = 7

        _shutdown_auto_index_executor()

        assert rag_module._pending_auto_index_jobs == 0

    def test_executor_has_bounded_workers(self):
        """Test that the executor has bounded max_workers to prevent thread proliferation."""
        from local_deep_research.research_library.routes.rag_routes import (
            _get_auto_index_executor,
        )

        executor = _get_auto_index_executor()
        # The executor should have max_workers=4 as configured
        assert executor._max_workers == 4

    def test_executor_submits_work_successfully(self):
        """Test that work can be submitted to the executor."""

        from local_deep_research.research_library.routes.rag_routes import (
            _get_auto_index_executor,
        )

        executor = _get_auto_index_executor()
        result = []

        def worker(value):
            result.append(value)

        future = executor.submit(worker, "test_value")
        future.result(timeout=5)  # Wait for completion

        assert result == ["test_value"]

    def test_executor_limits_concurrent_tasks(self):
        """Test that the executor limits concurrent tasks to max_workers."""
        from local_deep_research.research_library.routes.rag_routes import (
            _get_auto_index_executor,
        )

        executor = _get_auto_index_executor()
        max_workers = executor._max_workers

        running_count = []
        count_lock = threading.Lock()
        barrier = threading.Event()

        def slow_worker():
            with count_lock:
                running_count.append(1)
            barrier.wait(timeout=5)  # Wait until released
            with count_lock:
                running_count.pop()

        # Submit more tasks than max_workers
        futures = [executor.submit(slow_worker) for _ in range(max_workers + 2)]

        # Give threads time to start
        time.sleep(0.2)  # allow: unmarked-sleep

        # Only max_workers should be running
        with count_lock:
            concurrent_count = len(running_count)
        assert concurrent_count <= max_workers

        # Release all workers
        barrier.set()

        # Wait for all to complete
        for f in futures:
            f.result(timeout=5)

    def test_executor_thread_name_prefix(self):
        """Test that executor threads have the correct name prefix."""
        from local_deep_research.research_library.routes.rag_routes import (
            _get_auto_index_executor,
        )

        executor = _get_auto_index_executor()
        thread_name_captured = []

        def capture_thread_name():
            thread_name_captured.append(threading.current_thread().name)

        future = executor.submit(capture_thread_name)
        future.result(timeout=5)

        assert len(thread_name_captured) == 1
        assert thread_name_captured[0].startswith("auto_index_")

    def test_executor_thread_safe_initialization(self):
        """Test that executor initialization is thread-safe under concurrent access."""
        from local_deep_research.research_library.routes.rag_routes import (
            _get_auto_index_executor,
            _shutdown_auto_index_executor,
        )

        # Reset executor to test initialization
        _shutdown_auto_index_executor()

        executors = []
        errors = []
        barrier = threading.Barrier(10)

        def get_executor_concurrently():
            try:
                barrier.wait(timeout=5)  # Synchronize all threads
                executor = _get_auto_index_executor()
                executors.append(executor)
            except Exception as e:
                errors.append(e)

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

        assert len(errors) == 0, f"Errors occurred: {errors}"
        assert len(executors) == 10
        # All should be the same instance
        assert all(e is executors[0] for e in executors)

    def test_reserve_slot_never_oversubscribes_under_contention(self):
        """Concurrent reservations must never exceed the cap.

        ``_try_reserve_auto_index_slot`` guards the counter with
        ``_pending_auto_index_lock``. With the baseline set near the cap, only
        ``cap - baseline`` reservations may succeed no matter how many threads
        race. We fire N threads (N > the remaining capacity), each waiting on a
        ``threading.Barrier`` so they hit the critical section as close to
        simultaneously as possible, then assert:

        * the number of successful reservations equals exactly the remaining
          capacity (not more), and
        * the final counter never exceeds the cap.

        Mutation-awareness: if ``with _pending_auto_index_lock:`` is removed
        from ``_try_reserve_auto_index_slot``, two threads can both read the
        same pre-increment value and both succeed, over-subscribing past the
        cap and failing this test. Note this is *probabilistic* — the lost
        update only manifests on an actual interleaving — but the Barrier and a
        tight remaining capacity (1) maximize contention to make it reliable.
        """
        import local_deep_research.research_library.routes.rag_routes as rag_module
        from local_deep_research.research_library.routes.rag_routes import (
            _try_reserve_auto_index_slot,
        )

        cap = rag_module._MAX_PENDING_AUTO_INDEX_JOBS
        # Leave exactly ONE free slot so any lost update over-subscribes the
        # cap and is unambiguously detectable.
        remaining = 1
        baseline = cap - remaining

        n_threads = 16
        with rag_module._pending_auto_index_lock:
            rag_module._pending_auto_index_jobs = baseline

        barrier = threading.Barrier(n_threads)
        results = []
        results_lock = threading.Lock()

        def contend():
            barrier.wait(timeout=5)  # all threads collide here
            ok = _try_reserve_auto_index_slot()
            with results_lock:
                results.append(ok)

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

        successes = sum(1 for r in results if r)

        # Exactly the remaining capacity may be granted — never more.
        assert successes == remaining, (
            f"expected {remaining} successful reservation(s), got {successes}"
        )
        # The counter must never exceed the cap.
        assert rag_module._pending_auto_index_jobs <= cap
        # And it should land exactly at the cap (baseline + remaining grants).
        assert rag_module._pending_auto_index_jobs == baseline + remaining

    def test_executor_handles_worker_exceptions(self):
        """Test that worker exceptions don't break the executor."""
        from local_deep_research.research_library.routes.rag_routes import (
            _get_auto_index_executor,
        )

        executor = _get_auto_index_executor()

        def failing_worker():
            raise ValueError("Intentional test error")

        def succeeding_worker():
            return "success"

        # Submit a failing task
        future1 = executor.submit(failing_worker)

        # Submit a succeeding task after
        future2 = executor.submit(succeeding_worker)

        # First should raise
        try:
            future1.result(timeout=5)
            assert False, "Should have raised ValueError"
        except ValueError as e:
            assert "Intentional test error" in str(e)

        # Second should still succeed
        assert future2.result(timeout=5) == "success"

    def test_shutdown_is_idempotent(self):
        """Test that calling shutdown multiple times is safe."""
        import local_deep_research.research_library.routes.rag_routes as rag_module
        from local_deep_research.research_library.routes.rag_routes import (
            _get_auto_index_executor,
            _shutdown_auto_index_executor,
        )

        # Ensure executor exists
        _get_auto_index_executor()

        # Shutdown multiple times should not raise
        _shutdown_auto_index_executor()
        _shutdown_auto_index_executor()
        _shutdown_auto_index_executor()

        assert rag_module._auto_index_executor is None

    def test_shutdown_when_no_executor_exists(self):
        """Test that shutdown is safe when no executor has been created."""
        import local_deep_research.research_library.routes.rag_routes as rag_module
        from local_deep_research.research_library.routes.rag_routes import (
            _shutdown_auto_index_executor,
        )

        # Force executor to None
        rag_module._auto_index_executor = None

        # Should not raise
        _shutdown_auto_index_executor()

        assert rag_module._auto_index_executor is None

    def test_executor_queues_excess_tasks(self):
        """Test that tasks beyond max_workers are queued and eventually executed."""
        from local_deep_research.research_library.routes.rag_routes import (
            _get_auto_index_executor,
        )

        executor = _get_auto_index_executor()
        max_workers = executor._max_workers
        total_tasks = max_workers * 3  # Submit 3x more tasks than workers

        results = []
        results_lock = threading.Lock()

        def worker(task_id):
            time.sleep(0.05)  # Small delay
            with results_lock:
                results.append(task_id)
            return task_id

        # Submit many tasks
        futures = [executor.submit(worker, i) for i in range(total_tasks)]

        # Wait for all to complete
        for f in futures:
            f.result(timeout=30)

        # All tasks should have been executed
        assert len(results) == total_tasks
        assert set(results) == set(range(total_tasks))

    def test_executor_preserves_task_order_within_worker(self):
        """Test that a single worker processes its tasks in order."""
        from local_deep_research.research_library.routes.rag_routes import (
            _get_auto_index_executor,
        )

        executor = _get_auto_index_executor()
        results = []

        def worker(value):
            results.append(value)
            return value

        # Submit tasks sequentially and wait for each
        for i in range(5):
            future = executor.submit(worker, i)
            future.result(timeout=5)

        # Results should be in order since we waited for each
        assert results == [0, 1, 2, 3, 4]


class TestAutoIndexDocumentsWorker:
    """Test the _auto_index_documents_worker function."""

    def test_worker_indexes_documents(self):
        """Test that the worker indexes documents via RAG service."""
        from local_deep_research.research_library.routes.rag_routes import (
            _auto_index_documents_worker,
        )

        with patch(
            "local_deep_research.research_library.routes.rag_routes._get_rag_service_for_thread"
        ) as mock_get_service:
            mock_rag_service = MagicMock()
            mock_rag_service.index_document.return_value = {"status": "success"}
            # Configure context manager behavior
            mock_get_service.return_value.__enter__.return_value = (
                mock_rag_service
            )
            mock_get_service.return_value.__exit__.return_value = None

            _auto_index_documents_worker(
                document_ids=["doc1", "doc2", "doc3"],
                collection_id="coll1",
                username="testuser",
                db_password="testpass",
            )

            # Should have called index_document for each document
            assert mock_rag_service.index_document.call_count == 3
            mock_rag_service.index_document.assert_any_call(
                "doc1", "coll1", force_reindex=False
            )
            mock_rag_service.index_document.assert_any_call(
                "doc2", "coll1", force_reindex=False
            )
            mock_rag_service.index_document.assert_any_call(
                "doc3", "coll1", force_reindex=False
            )

    def test_worker_handles_skipped_documents(self):
        """Test that the worker handles already indexed documents."""
        from local_deep_research.research_library.routes.rag_routes import (
            _auto_index_documents_worker,
        )

        with patch(
            "local_deep_research.research_library.routes.rag_routes._get_rag_service_for_thread"
        ) as mock_get_service:
            mock_rag_service = MagicMock()
            # First succeeds, second skipped, third succeeds
            mock_rag_service.index_document.side_effect = [
                {"status": "success"},
                {"status": "skipped"},
                {"status": "success"},
            ]
            # Configure context manager behavior
            mock_get_service.return_value.__enter__.return_value = (
                mock_rag_service
            )
            mock_get_service.return_value.__exit__.return_value = None

            # Should not raise
            _auto_index_documents_worker(
                document_ids=["doc1", "doc2", "doc3"],
                collection_id="coll1",
                username="testuser",
                db_password="testpass",
            )

            # All documents should have been attempted
            assert mock_rag_service.index_document.call_count == 3

    def test_worker_handles_indexing_exception(self):
        """Test that the worker handles exceptions during indexing."""
        from local_deep_research.research_library.routes.rag_routes import (
            _auto_index_documents_worker,
        )

        with patch(
            "local_deep_research.research_library.routes.rag_routes._get_rag_service_for_thread"
        ) as mock_get_service:
            mock_rag_service = MagicMock()
            mock_rag_service.index_document.side_effect = Exception(
                "Index failed"
            )
            # Configure context manager behavior
            mock_get_service.return_value.__enter__.return_value = (
                mock_rag_service
            )
            mock_get_service.return_value.__exit__.return_value = None

            # Should not raise, even with exception
            _auto_index_documents_worker(
                document_ids=["doc1"],
                collection_id="coll1",
                username="testuser",
                db_password="testpass",
            )

    def test_worker_continues_after_exception(self):
        """Test that the worker continues indexing after an exception."""
        from local_deep_research.research_library.routes.rag_routes import (
            _auto_index_documents_worker,
        )

        with patch(
            "local_deep_research.research_library.routes.rag_routes._get_rag_service_for_thread"
        ) as mock_get_service:
            mock_rag_service = MagicMock()
            # First succeeds, second raises, third succeeds
            mock_rag_service.index_document.side_effect = [
                {"status": "success"},
                Exception("Index failed"),
                {"status": "success"},
            ]
            # Configure context manager behavior
            mock_get_service.return_value.__enter__.return_value = (
                mock_rag_service
            )
            mock_get_service.return_value.__exit__.return_value = None

            # Should not raise
            _auto_index_documents_worker(
                document_ids=["doc1", "doc2", "doc3"],
                collection_id="coll1",
                username="testuser",
                db_password="testpass",
            )

            # All documents should have been attempted
            assert mock_rag_service.index_document.call_count == 3

    def test_worker_creates_rag_service_with_correct_params(self):
        """Test that the worker creates RAG service with correct parameters."""
        from local_deep_research.research_library.routes.rag_routes import (
            _auto_index_documents_worker,
        )

        with patch(
            "local_deep_research.research_library.routes.rag_routes._get_rag_service_for_thread"
        ) as mock_get_service:
            mock_rag_service = MagicMock()
            mock_rag_service.index_document.return_value = {"status": "success"}
            # Configure context manager behavior
            mock_get_service.return_value.__enter__.return_value = (
                mock_rag_service
            )
            mock_get_service.return_value.__exit__.return_value = None

            _auto_index_documents_worker(
                document_ids=["doc1"],
                collection_id="my_collection",
                username="alice",
                db_password="secret",
            )

            mock_get_service.assert_called_once_with(
                "my_collection", "alice", "secret"
            )

    def test_worker_with_empty_document_list(self):
        """Test that the worker handles empty document list."""
        from local_deep_research.research_library.routes.rag_routes import (
            _auto_index_documents_worker,
        )

        with patch(
            "local_deep_research.research_library.routes.rag_routes._get_rag_service_for_thread"
        ) as mock_get_service:
            mock_rag_service = MagicMock()
            # Configure context manager behavior
            mock_get_service.return_value.__enter__.return_value = (
                mock_rag_service
            )
            mock_get_service.return_value.__exit__.return_value = None

            # Should not raise with empty list
            _auto_index_documents_worker(
                document_ids=[],
                collection_id="coll1",
                username="testuser",
                db_password="testpass",
            )

            # index_document should not be called
            mock_rag_service.index_document.assert_not_called()

    def test_worker_handles_rag_service_creation_failure(self):
        """Test that the worker handles RAG service creation failure."""
        from local_deep_research.research_library.routes.rag_routes import (
            _auto_index_documents_worker,
        )

        with patch(
            "local_deep_research.research_library.routes.rag_routes._get_rag_service_for_thread"
        ) as mock_get_service:
            mock_get_service.side_effect = Exception(
                "Failed to create RAG service"
            )

            # Should not raise, even with exception
            _auto_index_documents_worker(
                document_ids=["doc1"],
                collection_id="coll1",
                username="testuser",
                db_password="testpass",
            )


class TestAutoIndexIntegration:
    """Integration tests for the auto-indexing system."""

    def test_full_flow_with_real_executor(self):
        """Test the full flow using the real executor."""
        from local_deep_research.research_library.routes.rag_routes import (
            trigger_auto_index,
            _get_auto_index_executor,
        )

        # Ensure executor is initialized
        _get_auto_index_executor()
        task_executed = threading.Event()

        mock_rag_service = MagicMock()

        def index_and_signal(doc_id, collection_id, force_reindex=False):
            task_executed.set()
            return {"status": "success"}

        mock_rag_service.index_document.side_effect = index_and_signal

        with patch(
            "local_deep_research.database.session_context.get_user_db_session"
        ) as mock_session:
            mock_db = MagicMock()
            mock_session.return_value.__enter__.return_value = mock_db

            mock_settings = MagicMock()
            mock_settings.get_bool_setting.return_value = True

            with patch(
                "local_deep_research.research_library.routes.rag_routes.SettingsManager",
                return_value=mock_settings,
            ):
                with patch(
                    "local_deep_research.research_library.routes.rag_routes._get_rag_service_for_thread"
                ) as mock_get_service:
                    # Configure context manager behavior
                    mock_get_service.return_value.__enter__.return_value = (
                        mock_rag_service
                    )
                    mock_get_service.return_value.__exit__.return_value = None

                    trigger_auto_index(
                        document_ids=["doc1"],
                        collection_id="coll1",
                        username="testuser",
                        db_password="testpass",
                    )

                    # Wait for task to execute
                    assert task_executed.wait(timeout=5), (
                        "Task was not executed"
                    )

    def test_multiple_concurrent_triggers(self):
        """Test multiple concurrent trigger_auto_index calls."""
        from local_deep_research.research_library.routes.rag_routes import (
            trigger_auto_index,
            _get_auto_index_executor,
        )

        # Ensure executor is initialized
        _get_auto_index_executor()
        indexed_docs = []
        docs_lock = threading.Lock()
        all_done = threading.Event()
        expected_count = 5

        mock_rag_service = MagicMock()

        def track_indexing(doc_id, collection_id, force_reindex=False):
            with docs_lock:
                indexed_docs.append(doc_id)
                if len(indexed_docs) >= expected_count:
                    all_done.set()
            return {"status": "success"}

        mock_rag_service.index_document.side_effect = track_indexing

        with patch(
            "local_deep_research.database.session_context.get_user_db_session"
        ) as mock_session:
            mock_db = MagicMock()
            mock_session.return_value.__enter__.return_value = mock_db

            mock_settings = MagicMock()
            mock_settings.get_bool_setting.return_value = True

            with patch(
                "local_deep_research.research_library.routes.rag_routes.SettingsManager",
                return_value=mock_settings,
            ):
                with patch(
                    "local_deep_research.research_library.routes.rag_routes._get_rag_service_for_thread"
                ) as mock_get_service:
                    # Configure context manager behavior
                    mock_get_service.return_value.__enter__.return_value = (
                        mock_rag_service
                    )
                    mock_get_service.return_value.__exit__.return_value = None

                    # Trigger multiple times concurrently
                    for i in range(expected_count):
                        trigger_auto_index(
                            document_ids=[f"doc_{i}"],
                            collection_id="coll1",
                            username="testuser",
                            db_password="testpass",
                        )

                    # Wait for all to complete
                    assert all_done.wait(timeout=10), "Not all tasks completed"

                    with docs_lock:
                        assert len(indexed_docs) == expected_count

    def test_executor_reused_across_triggers(self):
        """Test that the same executor is reused across multiple triggers."""
        from local_deep_research.research_library.routes.rag_routes import (
            trigger_auto_index,
        )

        with patch(
            "local_deep_research.database.session_context.get_user_db_session"
        ) as mock_session:
            mock_db = MagicMock()
            mock_session.return_value.__enter__.return_value = mock_db

            mock_settings = MagicMock()
            mock_settings.get_bool_setting.return_value = True

            with patch(
                "local_deep_research.research_library.routes.rag_routes.SettingsManager",
                return_value=mock_settings,
            ):
                with patch(
                    "local_deep_research.research_library.routes.rag_routes._get_auto_index_executor"
                ) as mock_get_executor:
                    mock_executor = MagicMock()
                    mock_get_executor.return_value = mock_executor

                    # Trigger multiple times
                    for i in range(3):
                        trigger_auto_index(
                            document_ids=[f"doc_{i}"],
                            collection_id="coll1",
                            username="testuser",
                            db_password="testpass",
                        )

                    # Same executor should be fetched each time
                    assert mock_get_executor.call_count == 3
                    assert mock_executor.submit.call_count == 3