"""
Coverage tests for LibraryRAGService.

Focuses on logic paths not exercised by the existing test_library_rag_service.py:
- _get_index_hash edge cases
- _get_index_path cache directory details
- _deduplicate_chunks order preservation and empty input
- _get_or_create_rag_index new vs existing
- load_or_create_faiss_index HNSW/IVF/L2/IP variants, integrity failure, dimension mismatch,
  load failure, corrupted unlink failure
- index_document chunk indexing, empty text, skip already indexed
- index_all_documents with progress callback, stores settings
"""

import hashlib
import threading
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest
from langchain_core.documents import Document as LangchainDocument

# ---------------------------------------------------------------------------
# Module-level patch path prefix
# ---------------------------------------------------------------------------
_MOD = "local_deep_research.research_library.services.library_rag_service"


def _make_service(**overrides):
    """Create a LibraryRAGService with all external deps mocked out."""
    with (
        patch(f"{_MOD}.LocalEmbeddingManager") as _lem,
        patch(f"{_MOD}.get_user_db_session"),
        patch(f"{_MOD}.FileIntegrityManager") as _fim,
        patch(f"{_MOD}.get_text_splitter") as _gts,
    ):
        _lem.return_value.embeddings = MagicMock()
        from local_deep_research.research_library.services.library_rag_service import (
            LibraryRAGService,
        )

        defaults = dict(username="testuser", db_password="pw")
        defaults.update(overrides)
        svc = LibraryRAGService(**defaults)
    return svc


# =========================================================================
# _get_index_hash
# =========================================================================
class TestGetIndexHash:
    """Tests for _get_index_hash determinism and sensitivity."""

    def test_hash_is_deterministic(self):
        svc = _make_service()
        h1 = svc._get_index_hash("col_a", "model_x", "sentence_transformers")
        h2 = svc._get_index_hash("col_a", "model_x", "sentence_transformers")
        assert h1 == h2

    def test_hash_changes_with_collection_name(self):
        svc = _make_service()
        h1 = svc._get_index_hash("col_a", "model_x", "sentence_transformers")
        h2 = svc._get_index_hash("col_b", "model_x", "sentence_transformers")
        assert h1 != h2

    def test_hash_changes_with_model(self):
        svc = _make_service()
        h1 = svc._get_index_hash("col_a", "model_x", "sentence_transformers")
        h2 = svc._get_index_hash("col_a", "model_y", "sentence_transformers")
        assert h1 != h2

    def test_hash_changes_with_provider(self):
        svc = _make_service()
        h1 = svc._get_index_hash("col_a", "model_x", "sentence_transformers")
        h2 = svc._get_index_hash("col_a", "model_x", "ollama")
        assert h1 != h2

    def test_hash_is_sha256_hex(self):
        svc = _make_service()
        h = svc._get_index_hash("c", "m", "p")
        expected = hashlib.sha256("c:m:p".encode()).hexdigest()
        assert h == expected

    def test_hash_length_is_64_chars(self):
        svc = _make_service()
        h = svc._get_index_hash("x", "y", "z")
        assert len(h) == 64


# =========================================================================
# _get_index_path
# =========================================================================
class TestGetIndexPath:
    """Tests for _get_index_path."""

    @patch(f"{_MOD}.get_cache_directory")
    def test_path_under_rag_indices_subdir(self, mock_cache_dir, tmp_path):
        mock_cache_dir.return_value = tmp_path
        svc = _make_service()
        p = svc._get_index_path("abc123")
        assert p.parent.name == "rag_indices"

    @patch(f"{_MOD}.get_cache_directory")
    def test_path_filename_contains_hash(self, mock_cache_dir, tmp_path):
        mock_cache_dir.return_value = tmp_path
        svc = _make_service()
        p = svc._get_index_path("abc123")
        assert p.name == "abc123.faiss"

    @patch(f"{_MOD}.get_cache_directory")
    def test_path_creates_directory(self, mock_cache_dir, tmp_path):
        mock_cache_dir.return_value = tmp_path
        svc = _make_service()
        svc._get_index_path("abc123")
        assert (tmp_path / "rag_indices").is_dir()

    @patch(f"{_MOD}.get_cache_directory")
    def test_path_returns_path_object(self, mock_cache_dir, tmp_path):
        mock_cache_dir.return_value = tmp_path
        svc = _make_service()
        p = svc._get_index_path("somehash")
        assert isinstance(p, Path)


# =========================================================================
# _deduplicate_chunks  (static method)
# =========================================================================
class TestDeduplicateChunks:
    """Tests for _deduplicate_chunks."""

    def _doc(self, text):
        return LangchainDocument(page_content=text)

    def test_empty_input_returns_empty(self):
        from local_deep_research.research_library.services.library_rag_service import (
            LibraryRAGService,
        )

        chunks, ids = LibraryRAGService._deduplicate_chunks([], [])
        assert chunks == []
        assert ids == []

    def test_no_duplicates_preserved(self):
        from local_deep_research.research_library.services.library_rag_service import (
            LibraryRAGService,
        )

        d1, d2 = self._doc("a"), self._doc("b")
        chunks, ids = LibraryRAGService._deduplicate_chunks(
            [d1, d2], ["id1", "id2"]
        )
        assert ids == ["id1", "id2"]
        assert chunks == [d1, d2]

    def test_duplicate_ids_keeps_first(self):
        from local_deep_research.research_library.services.library_rag_service import (
            LibraryRAGService,
        )

        d1, d2, d3 = self._doc("a"), self._doc("b"), self._doc("c")
        chunks, ids = LibraryRAGService._deduplicate_chunks(
            [d1, d2, d3], ["id1", "id1", "id2"]
        )
        assert ids == ["id1", "id2"]
        assert chunks == [d1, d3]

    def test_existing_ids_excluded(self):
        from local_deep_research.research_library.services.library_rag_service import (
            LibraryRAGService,
        )

        d1, d2 = self._doc("a"), self._doc("b")
        chunks, ids = LibraryRAGService._deduplicate_chunks(
            [d1, d2], ["id1", "id2"], existing_ids={"id1"}
        )
        assert ids == ["id2"]
        assert chunks == [d2]

    def test_none_existing_ids_allows_all(self):
        from local_deep_research.research_library.services.library_rag_service import (
            LibraryRAGService,
        )

        d1 = self._doc("a")
        chunks, ids = LibraryRAGService._deduplicate_chunks(
            [d1], ["id1"], existing_ids=None
        )
        assert ids == ["id1"]

    def test_order_preservation(self):
        from local_deep_research.research_library.services.library_rag_service import (
            LibraryRAGService,
        )

        docs = [self._doc(str(i)) for i in range(5)]
        id_list = ["e", "d", "c", "b", "a"]
        chunks, ids = LibraryRAGService._deduplicate_chunks(docs, id_list)
        assert ids == ["e", "d", "c", "b", "a"]

    def test_all_existing_returns_empty(self):
        from local_deep_research.research_library.services.library_rag_service import (
            LibraryRAGService,
        )

        d1, d2 = self._doc("a"), self._doc("b")
        chunks, ids = LibraryRAGService._deduplicate_chunks(
            [d1, d2], ["id1", "id2"], existing_ids={"id1", "id2"}
        )
        assert chunks == []
        assert ids == []


# =========================================================================
# _get_or_create_rag_index
# =========================================================================
class TestGetOrCreateRagIndex:
    """Tests for _get_or_create_rag_index."""

    @patch(f"{_MOD}.get_user_db_session")
    def test_returns_existing_index(self, mock_session_ctx):
        svc = _make_service()
        mock_session = MagicMock()
        mock_session_ctx.return_value.__enter__ = MagicMock(
            return_value=mock_session
        )
        mock_session_ctx.return_value.__exit__ = MagicMock(return_value=None)

        existing_index = MagicMock()
        mock_session.query.return_value.filter_by.return_value.first.return_value = existing_index

        result = svc._get_or_create_rag_index("coll-123")
        assert result is existing_index
        # Should NOT call session.add since index already existed
        mock_session.add.assert_not_called()

    @patch(f"{_MOD}.get_cache_directory")
    @patch(f"{_MOD}.get_user_db_session")
    def test_creates_new_index_when_none_exists(
        self, mock_session_ctx, mock_cache_dir, tmp_path
    ):
        mock_cache_dir.return_value = tmp_path
        svc = _make_service()
        svc.embedding_manager = MagicMock()
        svc.embedding_manager.embeddings.embed_query.return_value = [0.0] * 384

        mock_session = MagicMock()
        mock_session_ctx.return_value.__enter__ = MagicMock(
            return_value=mock_session
        )
        mock_session_ctx.return_value.__exit__ = MagicMock(return_value=None)

        # No existing index found
        mock_session.query.return_value.filter_by.return_value.first.return_value = None

        svc._get_or_create_rag_index("coll-456")

        # Should have called session.add for the new RAGIndex
        mock_session.add.assert_called_once()
        mock_session.commit.assert_called_once()
        mock_session.refresh.assert_called_once()

    @patch(f"{_MOD}.get_cache_directory")
    @patch(f"{_MOD}.get_user_db_session")
    def test_embeds_test_string_for_dimension(
        self, mock_session_ctx, mock_cache_dir, tmp_path
    ):
        mock_cache_dir.return_value = tmp_path
        svc = _make_service()
        svc.embedding_manager = MagicMock()
        svc.embedding_manager.embeddings.embed_query.return_value = [0.1] * 768

        mock_session = MagicMock()
        mock_session_ctx.return_value.__enter__ = MagicMock(
            return_value=mock_session
        )
        mock_session_ctx.return_value.__exit__ = MagicMock(return_value=None)
        mock_session.query.return_value.filter_by.return_value.first.return_value = None

        svc._get_or_create_rag_index("coll-789")

        svc.embedding_manager.embeddings.embed_query.assert_called_once_with(
            "test"
        )


# =========================================================================
# load_or_create_faiss_index
# =========================================================================
class TestLoadOrCreateFaissIndex:
    """Tests for load_or_create_faiss_index."""

    def _patch_get_or_create(self, svc, rag_index):
        """Patch _get_or_create_rag_index on a service instance."""
        svc._get_or_create_rag_index = MagicMock(return_value=rag_index)

    def _make_rag_index(self, index_path="/tmp/test.faiss", dim=384):
        idx = MagicMock()
        idx.index_path = index_path
        idx.embedding_dimension = dim
        idx.id = "rag-idx-1"
        return idx

    @patch(f"{_MOD}.FAISS")
    @patch(f"{_MOD}.InMemoryDocstore")
    @patch(f"{_MOD}.IndexFlatIP")
    def test_creates_flat_ip_for_cosine(
        self, mock_flat_ip, mock_docstore, mock_faiss
    ):
        svc = _make_service(distance_metric="cosine", index_type="flat")
        rag_idx = self._make_rag_index(index_path="/nonexistent/test.faiss")
        self._patch_get_or_create(svc, rag_idx)
        svc.embedding_manager = MagicMock()

        svc.load_or_create_faiss_index("coll-1")
        mock_flat_ip.assert_called_once_with(384)

    @patch(f"{_MOD}.FAISS")
    @patch(f"{_MOD}.InMemoryDocstore")
    @patch(f"{_MOD}.IndexFlatL2")
    def test_creates_flat_l2_for_l2_metric(
        self, mock_flat_l2, mock_docstore, mock_faiss
    ):
        svc = _make_service(distance_metric="l2", index_type="flat")
        rag_idx = self._make_rag_index(index_path="/nonexistent/test.faiss")
        self._patch_get_or_create(svc, rag_idx)
        svc.embedding_manager = MagicMock()

        svc.load_or_create_faiss_index("coll-1")
        mock_flat_l2.assert_called_once_with(384)

    @patch(f"{_MOD}.FAISS")
    @patch(f"{_MOD}.InMemoryDocstore")
    @patch(f"{_MOD}.IndexHNSWFlat")
    def test_creates_hnsw_index(self, mock_hnsw, mock_docstore, mock_faiss):
        svc = _make_service(index_type="hnsw")
        rag_idx = self._make_rag_index(index_path="/nonexistent/test.faiss")
        self._patch_get_or_create(svc, rag_idx)
        svc.embedding_manager = MagicMock()

        svc.load_or_create_faiss_index("coll-1")
        mock_hnsw.assert_called_once_with(384, 32)

    @patch(f"{_MOD}.FAISS")
    @patch(f"{_MOD}.InMemoryDocstore")
    @patch(f"{_MOD}.IndexFlatIP")
    def test_ivf_falls_back_to_flat_ip_for_cosine(
        self, mock_flat_ip, mock_docstore, mock_faiss
    ):
        svc = _make_service(index_type="ivf", distance_metric="cosine")
        rag_idx = self._make_rag_index(index_path="/nonexistent/test.faiss")
        self._patch_get_or_create(svc, rag_idx)
        svc.embedding_manager = MagicMock()

        svc.load_or_create_faiss_index("coll-1")
        mock_flat_ip.assert_called_once_with(384)

    @patch(f"{_MOD}.FAISS")
    @patch(f"{_MOD}.InMemoryDocstore")
    @patch(f"{_MOD}.IndexFlatL2")
    def test_ivf_falls_back_to_flat_l2_for_l2(
        self, mock_flat_l2, mock_docstore, mock_faiss
    ):
        svc = _make_service(index_type="ivf", distance_metric="l2")
        rag_idx = self._make_rag_index(index_path="/nonexistent/test.faiss")
        self._patch_get_or_create(svc, rag_idx)
        svc.embedding_manager = MagicMock()

        svc.load_or_create_faiss_index("coll-1")
        mock_flat_l2.assert_called_once_with(384)

    @patch(f"{_MOD}.safe_load_faiss")
    @patch(f"{_MOD}.FAISS")
    @patch(f"{_MOD}.InMemoryDocstore")
    @patch(f"{_MOD}.IndexFlatIP")
    def test_verified_load_returns_existing_index(
        self, mock_flat_ip, mock_docstore, mock_faiss_cls, mock_safe_load
    ):
        svc = _make_service()
        svc.embedding_manager = MagicMock()
        svc.embedding_manager.embeddings.embed_query.return_value = [0.0] * 384
        svc.integrity_manager = MagicMock()
        svc.integrity_manager.verify_file.return_value = (True, None)

        rag_idx = self._make_rag_index(dim=384)
        rag_idx.index_path = "/tmp/existing.faiss"
        self._patch_get_or_create(svc, rag_idx)

        mock_loaded = MagicMock()
        mock_safe_load.return_value = mock_loaded

        with patch.object(Path, "exists", return_value=True):
            result = svc.load_or_create_faiss_index("coll-1")

        # Verified index is loaded via the restricted-unpickler loader,
        # never via the dangerous FAISS.load_local.
        assert result is mock_loaded
        mock_safe_load.assert_called_once()
        mock_faiss_cls.load_local.assert_not_called()

    @patch(f"{_MOD}.FAISS")
    @patch(f"{_MOD}.InMemoryDocstore")
    @patch(f"{_MOD}.IndexFlatIP")
    def test_integrity_failure_quarantines_and_creates_new_index(
        self, mock_flat_ip, mock_docstore, mock_faiss_cls, tmp_path
    ):
        """Verify-failure path: corrupt .faiss + .pkl are RENAMED to
        .corrupt-<ns>, NOT unlinked. Then fresh index is created.
        Regression for #4197 data-loss bug.
        """
        svc = _make_service()
        svc.embedding_manager = MagicMock()
        svc.integrity_manager = MagicMock()
        svc.integrity_manager.verify_file.return_value = (
            False,
            "hash mismatch",
        )

        # Use real on-disk files so .exists() in the quarantine
        # collision check returns False naturally for the .corrupt-*
        # paths, not True-for-everything.
        idx_path = tmp_path / "corrupt.faiss"
        pkl_path = tmp_path / "corrupt.pkl"
        idx_path.write_bytes(b"faiss-bytes")
        pkl_path.write_bytes(b"pkl-bytes")

        rag_idx = self._make_rag_index()
        rag_idx.index_path = str(idx_path)
        self._patch_get_or_create(svc, rag_idx)

        with patch.object(Path, "unlink") as mock_unlink:
            svc.load_or_create_faiss_index("coll-1")

        # Should NOT have unlinked anything in the verify-failure path
        mock_unlink.assert_not_called()
        # Originals quarantined (renamed away from their paths)
        assert not idx_path.exists()
        assert not pkl_path.exists()
        # Both files preserved under .corrupt-<ns> names
        corrupt_faiss = list(tmp_path.glob("corrupt.faiss.corrupt-*"))
        corrupt_pkl = list(tmp_path.glob("corrupt.pkl.corrupt-*"))
        assert len(corrupt_faiss) == 1
        assert len(corrupt_pkl) == 1
        assert corrupt_faiss[0].read_bytes() == b"faiss-bytes"
        # Should return a new FAISS instance, not load_local
        mock_faiss_cls.load_local.assert_not_called()
        mock_faiss_cls.assert_called_once()

    @patch(f"{_MOD}.FAISS")
    @patch(f"{_MOD}.InMemoryDocstore")
    @patch(f"{_MOD}.IndexFlatIP")
    def test_quarantine_rename_oserror_re_raises(
        self, mock_flat_ip, mock_docstore, mock_faiss_cls, tmp_path
    ):
        """Disk-full / read-only fs during quarantine MUST propagate.
        Silently falling through would let the next save_local truncate
        the corrupt bytes, recreating the very data loss #4197 fixes.
        """
        svc = _make_service()
        svc.embedding_manager = MagicMock()
        svc.integrity_manager = MagicMock()
        svc.integrity_manager.verify_file.return_value = (False, "bad hash")

        idx_path = tmp_path / "corrupt.faiss"
        idx_path.write_bytes(b"faiss-bytes")
        (tmp_path / "corrupt.pkl").write_bytes(b"pkl-bytes")

        rag_idx = self._make_rag_index()
        rag_idx.index_path = str(idx_path)
        self._patch_get_or_create(svc, rag_idx)

        with patch.object(Path, "rename", side_effect=OSError("ENOSPC")):
            with pytest.raises(OSError, match="ENOSPC"):
                svc.load_or_create_faiss_index("coll-1")

        # Fresh index must NOT have been created — corrupt bytes are
        # still on disk and creating one would let the next save
        # overwrite them.
        mock_faiss_cls.assert_not_called()

    @patch(f"{_MOD}.get_user_db_session")
    @patch(f"{_MOD}.FAISS")
    @patch(f"{_MOD}.InMemoryDocstore")
    @patch(f"{_MOD}.IndexFlatIP")
    def test_dimension_mismatch_deletes_and_rebuilds(
        self, mock_flat_ip, mock_docstore, mock_faiss_cls, mock_session_ctx
    ):
        svc = _make_service()
        svc.embedding_manager = MagicMock()
        # Current model returns dim 768 but index was stored with 384
        svc.embedding_manager.embeddings.embed_query.return_value = [0.0] * 768
        svc.integrity_manager = MagicMock()
        svc.integrity_manager.verify_file.return_value = (True, None)

        mock_session = MagicMock()
        mock_session_ctx.return_value.__enter__ = MagicMock(
            return_value=mock_session
        )
        mock_session_ctx.return_value.__exit__ = MagicMock(return_value=None)
        mock_session.query.return_value.filter_by.return_value.first.return_value = MagicMock()

        rag_idx = self._make_rag_index(dim=384)
        rag_idx.index_path = "/tmp/old_dim.faiss"
        self._patch_get_or_create(svc, rag_idx)

        with (
            patch.object(Path, "exists", return_value=True),
            patch.object(Path, "unlink") as mock_unlink,
            patch.object(Path, "with_suffix") as mock_with_suffix,
        ):
            mock_pkl = MagicMock()
            mock_pkl.exists.return_value = True
            mock_with_suffix.return_value = mock_pkl
            svc.load_or_create_faiss_index("coll-1")

        # The old file should have been unlinked
        mock_unlink.assert_called()
        # A new FAISS should be created (not loaded)
        mock_faiss_cls.assert_called_once()

    @patch(f"{_MOD}.safe_load_faiss")
    @patch(f"{_MOD}.FAISS")
    @patch(f"{_MOD}.InMemoryDocstore")
    @patch(f"{_MOD}.IndexFlatIP")
    def test_load_failure_quarantines_and_creates_new_index(
        self,
        mock_flat_ip,
        mock_docstore,
        mock_faiss_cls,
        mock_safe_load,
        tmp_path,
    ):
        """When the loader raises (torn .pkl, malformed pickle, or a
        rejected/tampered pickle), the .faiss and .pkl are quarantined
        before falling through to a fresh index. Previously the old code
        silently discarded the broken-state files without preserving
        evidence.
        """
        svc = _make_service()
        svc.embedding_manager = MagicMock()
        svc.embedding_manager.embeddings.embed_query.return_value = [0.0] * 384
        svc.integrity_manager = MagicMock()
        svc.integrity_manager.verify_file.return_value = (True, None)

        idx_path = tmp_path / "broken.faiss"
        pkl_path = tmp_path / "broken.pkl"
        idx_path.write_bytes(b"faiss-bytes")
        pkl_path.write_bytes(b"pkl-bytes")

        rag_idx = self._make_rag_index(dim=384)
        rag_idx.index_path = str(idx_path)
        self._patch_get_or_create(svc, rag_idx)

        mock_safe_load.side_effect = RuntimeError("corrupted file")

        svc.load_or_create_faiss_index("coll-1")

        # Both files quarantined (renamed away)
        assert not idx_path.exists()
        assert not pkl_path.exists()
        assert len(list(tmp_path.glob("broken.faiss.corrupt-*"))) == 1
        assert len(list(tmp_path.glob("broken.pkl.corrupt-*"))) == 1
        # Should fall through and create new index
        mock_faiss_cls.assert_called_once()

    @patch(f"{_MOD}.FAISS")
    @patch(f"{_MOD}.InMemoryDocstore")
    @patch(f"{_MOD}.IndexFlatIP")
    def test_embedding_provider_failure_raises_and_preserves_index(
        self, mock_flat_ip, mock_docstore, mock_faiss_cls, tmp_path
    ):
        """When the embedding provider is unreachable (e.g. Ollama down),
        the dimension-check embed_query raises. That says nothing about
        the index files, so the error must propagate WITHOUT quarantining
        the healthy index or replacing it with an empty one.
        """
        svc = _make_service()
        svc.embedding_manager = MagicMock()
        svc.embedding_manager.embeddings.embed_query.side_effect = (
            ConnectionError("Ollama connection refused")
        )
        svc.integrity_manager = MagicMock()
        svc.integrity_manager.verify_file.return_value = (True, None)

        idx_path = tmp_path / "healthy.faiss"
        pkl_path = tmp_path / "healthy.pkl"
        idx_path.write_bytes(b"faiss-bytes")
        pkl_path.write_bytes(b"pkl-bytes")

        rag_idx = self._make_rag_index(dim=384)
        rag_idx.index_path = str(idx_path)
        self._patch_get_or_create(svc, rag_idx)

        with pytest.raises(ConnectionError, match="Ollama"):
            svc.load_or_create_faiss_index("coll-1")

        # The healthy index files must be untouched — not quarantined
        assert idx_path.read_bytes() == b"faiss-bytes"
        assert pkl_path.read_bytes() == b"pkl-bytes"
        assert list(tmp_path.glob("*.corrupt-*")) == []
        # And no empty replacement index was created
        mock_faiss_cls.load_local.assert_not_called()
        mock_faiss_cls.assert_not_called()

    @patch(f"{_MOD}.FAISS")
    @patch(f"{_MOD}.InMemoryDocstore")
    @patch(f"{_MOD}.IndexFlatIP")
    def test_normalize_vectors_passed_to_faiss(
        self, mock_flat_ip, mock_docstore, mock_faiss_cls
    ):
        svc = _make_service(normalize_vectors=False)
        rag_idx = self._make_rag_index(index_path="/nonexistent/test.faiss")
        self._patch_get_or_create(svc, rag_idx)
        svc.embedding_manager = MagicMock()

        svc.load_or_create_faiss_index("coll-1")
        call_kwargs = mock_faiss_cls.call_args
        assert call_kwargs[1]["normalize_L2"] is False


# =========================================================================
# index_document
# =========================================================================
class TestIndexDocument:
    """Tests for index_document."""

    @patch(f"{_MOD}.get_user_db_session")
    def test_returns_error_when_document_not_found(self, mock_session_ctx):
        svc = _make_service()
        mock_session = MagicMock()
        mock_session_ctx.return_value.__enter__ = MagicMock(
            return_value=mock_session
        )
        mock_session_ctx.return_value.__exit__ = MagicMock(return_value=None)
        mock_session.query.return_value.filter_by.return_value.first.return_value = None

        result = svc.index_document("doc-1", "coll-1")
        assert result["status"] == "error"
        assert "not found" in result["error"]

    @patch(f"{_MOD}.ensure_in_collection")
    @patch(f"{_MOD}.get_user_db_session")
    def test_returns_error_when_no_text_content(
        self, mock_session_ctx, mock_ensure
    ):
        svc = _make_service()
        mock_session = MagicMock()
        mock_session_ctx.return_value.__enter__ = MagicMock(
            return_value=mock_session
        )
        mock_session_ctx.return_value.__exit__ = MagicMock(return_value=None)

        mock_document = MagicMock()
        mock_document.text_content = None

        mock_session.query.return_value.filter_by.return_value.first.return_value = mock_document

        mock_ensure.return_value = MagicMock(indexed=False, chunk_count=0)

        result = svc.index_document("doc-1", "coll-1")
        assert result["status"] == "error"
        assert "no text content" in result["error"]

    @patch(f"{_MOD}.ensure_in_collection")
    @patch(f"{_MOD}.get_user_db_session")
    def test_skips_already_indexed_document(
        self, mock_session_ctx, mock_ensure
    ):
        svc = _make_service()
        mock_session = MagicMock()
        mock_session_ctx.return_value.__enter__ = MagicMock(
            return_value=mock_session
        )
        mock_session_ctx.return_value.__exit__ = MagicMock(return_value=None)

        mock_document = MagicMock()
        mock_session.query.return_value.filter_by.return_value.first.return_value = mock_document

        mock_ensure.return_value = MagicMock(indexed=True, chunk_count=42)

        result = svc.index_document("doc-1", "coll-1", force_reindex=False)
        assert result["status"] == "skipped"
        assert result["chunk_count"] == 42


# =========================================================================
# index_all_documents (index_collection in the spec)
# =========================================================================
class TestIndexAllDocuments:
    """Tests for index_all_documents."""

    @patch(f"{_MOD}.get_user_db_session")
    def test_no_documents_returns_info(self, mock_session_ctx):
        svc = _make_service()
        mock_session = MagicMock()
        mock_session_ctx.return_value.__enter__ = MagicMock(
            return_value=mock_session
        )
        mock_session_ctx.return_value.__exit__ = MagicMock(return_value=None)
        mock_session.query.return_value.filter_by.return_value.filter_by.return_value.all.return_value = []

        result = svc.index_all_documents("coll-1")
        assert result["status"] == "info"
        assert result["successful"] == 0

    @patch(f"{_MOD}.get_user_db_session")
    def test_progress_callback_invoked(self, mock_session_ctx):
        svc = _make_service()
        mock_session = MagicMock()
        mock_session_ctx.return_value.__enter__ = MagicMock(
            return_value=mock_session
        )
        mock_session_ctx.return_value.__exit__ = MagicMock(return_value=None)

        mock_dc1 = MagicMock()
        mock_dc1.document_id = "doc-1"
        mock_dc2 = MagicMock()
        mock_dc2.document_id = "doc-2"

        mock_doc = MagicMock()
        mock_doc.title = "Test Doc"

        # filter_by(collection_id=...) -> filter_by(indexed=False) -> all()
        mock_session.query.return_value.filter_by.return_value.filter_by.return_value.all.return_value = [
            mock_dc1,
            mock_dc2,
        ]
        # query(Document).filter_by(id=...).first() for title lookup
        mock_session.query.return_value.filter_by.return_value.first.return_value = mock_doc

        # Mock index_document to return success
        svc.index_document = MagicMock(
            return_value={"status": "success", "chunk_count": 10}
        )

        callback = MagicMock()
        svc.index_all_documents("coll-1", progress_callback=callback)

        assert callback.call_count == 2
        # Check the callback was called with (idx, total, title, status)
        first_call = callback.call_args_list[0]
        assert first_call[0][0] == 1  # idx
        assert first_call[0][1] == 2  # total
        assert first_call[0][3] == "success"  # status

    @patch(f"{_MOD}.get_user_db_session")
    def test_counts_successful_skipped_failed(self, mock_session_ctx):
        svc = _make_service()
        mock_session = MagicMock()
        mock_session_ctx.return_value.__enter__ = MagicMock(
            return_value=mock_session
        )
        mock_session_ctx.return_value.__exit__ = MagicMock(return_value=None)

        mock_dc1 = MagicMock()
        mock_dc1.document_id = "doc-1"
        mock_dc2 = MagicMock()
        mock_dc2.document_id = "doc-2"
        mock_dc3 = MagicMock()
        mock_dc3.document_id = "doc-3"

        mock_doc = MagicMock()
        mock_doc.title = "Title"

        # With force_reindex=True the code does:
        #   query(DocumentCollection).filter_by(collection_id=...).all()
        # (no second filter_by for indexed=False)
        mock_query = MagicMock()
        mock_query.filter_by.return_value.all.return_value = [
            mock_dc1,
            mock_dc2,
            mock_dc3,
        ]
        # Document title lookup
        mock_query.filter_by.return_value.first.return_value = mock_doc
        mock_session.query.return_value = mock_query

        results_sequence = [
            {"status": "success", "chunk_count": 5},
            {
                "status": "skipped",
                "message": "already indexed",
                "chunk_count": 3,
            },
            {"status": "error", "error": "something broke"},
        ]
        svc.index_document = MagicMock(side_effect=results_sequence)

        result = svc.index_all_documents("coll-1", force_reindex=True)
        assert result["successful"] == 1
        assert result["skipped"] == 1
        assert result["failed"] == 1
        assert len(result["errors"]) == 1


# =========================================================================
# db_password property
# =========================================================================
class TestDbPasswordProperty:
    """Tests for the db_password property propagation."""

    def test_getter_returns_value(self):
        svc = _make_service()
        svc._db_password = "secret"
        assert svc.db_password == "secret"

    def test_setter_propagates_to_embedding_manager(self):
        svc = _make_service()
        svc.embedding_manager = MagicMock()
        svc.integrity_manager = MagicMock()
        svc.db_password = "new_pw"
        assert svc.embedding_manager.db_password == "new_pw"

    def test_setter_propagates_to_integrity_manager(self):
        svc = _make_service()
        svc.embedding_manager = MagicMock()
        svc.integrity_manager = MagicMock()
        svc.db_password = "new_pw"
        assert svc.integrity_manager.password == "new_pw"

    def test_setter_handles_none_managers(self):
        svc = _make_service()
        svc.embedding_manager = None
        svc.integrity_manager = None
        # Should not raise
        svc.db_password = "pw"
        assert svc._db_password == "pw"


# =========================================================================
# Context manager / close
# =========================================================================
class TestContextManager:
    """Tests for context manager protocol."""

    def test_enter_returns_self(self):
        svc = _make_service()
        assert svc.__enter__() is svc

    def test_exit_calls_close(self):
        svc = _make_service()
        svc.close = MagicMock()
        svc.__exit__(None, None, None)
        svc.close.assert_called_once()

    def test_exit_returns_false(self):
        svc = _make_service()
        result = svc.__exit__(None, None, None)
        assert result is False


# =========================================================================
# Quarantine helper + concurrent-write locks (#4197)
# =========================================================================


def _import_module():
    import local_deep_research.research_library.services.library_rag_service as m

    return m


class TestCorruptionQuarantine:
    """Verify the verify-failure and load-failure paths preserve
    corrupted bytes instead of deleting them.
    """

    def _make_idx(self, tmp_path, name="idx"):
        idx = tmp_path / f"{name}.faiss"
        pkl = tmp_path / f"{name}.pkl"
        idx.write_bytes(b"faiss-bytes")
        pkl.write_bytes(b"pkl-bytes")
        return idx, pkl

    def test_renames_both_faiss_and_pkl(self, tmp_path):
        svc = _make_service()
        idx, pkl = self._make_idx(tmp_path)
        svc._quarantine_corrupt_index(idx, "test")
        # Original files are gone from their paths
        assert not idx.exists()
        assert not pkl.exists()
        # Both quarantined as .corrupt-*
        corrupted_faiss = list(tmp_path.glob("idx.faiss.corrupt-*"))
        corrupted_pkl = list(tmp_path.glob("idx.pkl.corrupt-*"))
        assert len(corrupted_faiss) == 1
        assert len(corrupted_pkl) == 1
        assert corrupted_faiss[0].read_bytes() == b"faiss-bytes"
        assert corrupted_pkl[0].read_bytes() == b"pkl-bytes"

    def test_skips_pkl_when_missing(self, tmp_path):
        svc = _make_service()
        idx = tmp_path / "idx.faiss"
        idx.write_bytes(b"faiss-bytes")
        # No .pkl present
        svc._quarantine_corrupt_index(idx, "test")
        assert not idx.exists()
        corrupted = list(tmp_path.glob("idx.faiss.corrupt-*"))
        assert len(corrupted) == 1
        # No orphan pkl created
        assert not any(tmp_path.glob("idx.pkl.corrupt-*"))

    def test_rename_oserror_re_raises(self, tmp_path):
        """Disk-full / permission failure must propagate, not be swallowed."""
        svc = _make_service()
        idx, _pkl = self._make_idx(tmp_path)
        with patch.object(Path, "rename", side_effect=OSError("EROFS")):
            with pytest.raises(OSError, match="EROFS"):
                svc._quarantine_corrupt_index(idx, "test")

    def test_collision_increments_suffix(self, tmp_path):
        """Pre-existing quarantine path → loop increments to -1, -2, ..."""
        svc = _make_service()
        idx, pkl = self._make_idx(tmp_path)
        # Force a specific timestamp so we can predict collision targets
        with patch.object(
            _import_module().time,
            "time_ns",
            return_value=12345,
        ):
            # Pre-create the base collision target
            (tmp_path / "idx.faiss.corrupt-12345").write_bytes(b"old1")
            svc._quarantine_corrupt_index(idx, "test")
        # Original moved out
        assert not idx.exists()
        # New corrupt-12345-1 created (the -1 increment)
        moved = tmp_path / "idx.faiss.corrupt-12345-1"
        assert moved.exists()
        assert moved.read_bytes() == b"faiss-bytes"

    def test_collision_cap_raises(self, tmp_path):
        """If 32 collisions in a row, surface an OSError, don't hang."""
        svc = _make_service()
        mod = _import_module()
        idx, pkl = self._make_idx(tmp_path)
        with patch.object(mod.time, "time_ns", return_value=999):
            # Pre-create base + 32 collisions
            (tmp_path / "idx.faiss.corrupt-999").write_bytes(b"")
            for n in range(1, mod._QUARANTINE_SUFFIX_RETRY_CAP + 1):
                (tmp_path / f"idx.faiss.corrupt-999-{n}").write_bytes(b"")
            with pytest.raises(OSError, match="collisions exceeded"):
                svc._quarantine_corrupt_index(idx, "test")


class TestQuarantineRetention:
    """Retention sweep keeps the rag_indices/ directory from filling
    with old .corrupt-* files on systems that experience recurring
    corruption.
    """

    def _make_idx(self, tmp_path, name="idx"):
        idx = tmp_path / f"{name}.faiss"
        pkl = tmp_path / f"{name}.pkl"
        idx.write_bytes(b"faiss-bytes")
        pkl.write_bytes(b"pkl-bytes")
        return idx, pkl

    def _make_quarantined(self, tmp_path, name, ns, mtime):
        """Create a paired (.faiss, .pkl) quarantined-style file at
        the given ns suffix and stamp it with ``mtime`` so the
        retention sweep's age sort is deterministic.
        """
        faiss_f = tmp_path / f"{name}.faiss.corrupt-{ns}"
        pkl_f = tmp_path / f"{name}.pkl.corrupt-{ns}"
        faiss_f.write_bytes(b"")
        pkl_f.write_bytes(b"")
        import os

        os.utime(faiss_f, (mtime, mtime))
        os.utime(pkl_f, (mtime, mtime))
        return faiss_f, pkl_f

    def test_prune_keeps_most_recent_drops_older(self, tmp_path):
        """After N+2 simulated past corruption events, pruning leaves
        exactly N (most recent by mtime) and removes the oldest 2.
        """
        svc = _make_service()
        mod = _import_module()
        keep = mod._QUARANTINE_KEEP_RECENT

        # Pre-seed keep+2 quarantined pairs with increasing mtimes
        # (later ns → newer mtime).
        for i in range(keep + 2):
            self._make_quarantined(
                tmp_path, "idx", ns=1000 + i, mtime=1000.0 + i
            )

        # Trigger pruning via the public quarantine call (which now
        # invokes the sweep at the end).
        idx, _pkl = self._make_idx(tmp_path)
        svc._quarantine_corrupt_index(idx, "test")

        faiss_corrupts = sorted(tmp_path.glob("idx.faiss.corrupt-*"))
        pkl_corrupts = sorted(tmp_path.glob("idx.pkl.corrupt-*"))
        # keep pre-existing + 1 fresh = keep total on each side
        assert len(faiss_corrupts) == keep
        assert len(pkl_corrupts) == keep

        # Oldest two pre-existing pairs (ns 1000, 1001) must be gone.
        for stale_ns in (1000, 1001):
            assert not (tmp_path / f"idx.faiss.corrupt-{stale_ns}").exists()
            assert not (tmp_path / f"idx.pkl.corrupt-{stale_ns}").exists()

    def test_prune_skipped_when_under_threshold(self, tmp_path):
        """With fewer than the retention cap, no pruning happens."""
        svc = _make_service()
        mod = _import_module()
        keep = mod._QUARANTINE_KEEP_RECENT

        # Pre-seed (keep - 2) quarantined pairs. After this call we'll
        # add 1 more via the quarantine, ending at keep - 1 — still
        # under the cap.
        for i in range(keep - 2):
            self._make_quarantined(
                tmp_path, "idx", ns=2000 + i, mtime=2000.0 + i
            )

        idx, _pkl = self._make_idx(tmp_path)
        svc._quarantine_corrupt_index(idx, "test")

        # All pre-existing + the new one survive.
        assert len(list(tmp_path.glob("idx.faiss.corrupt-*"))) == keep - 1

    def test_prune_only_targets_same_base(self, tmp_path):
        """Pruning of ``idx.faiss.corrupt-*`` must NOT touch
        ``other.faiss.corrupt-*`` — different base paths are
        independent.
        """
        svc = _make_service()
        mod = _import_module()
        keep = mod._QUARANTINE_KEEP_RECENT

        # Many old corrupts for the other base — these MUST be untouched.
        for i in range(keep + 3):
            self._make_quarantined(
                tmp_path, "other", ns=3000 + i, mtime=3000.0 + i
            )
        # Trigger many corrupts for our base so the sweep runs.
        for i in range(keep + 2):
            self._make_quarantined(
                tmp_path, "idx", ns=4000 + i, mtime=4000.0 + i
            )

        idx, _pkl = self._make_idx(tmp_path)
        svc._quarantine_corrupt_index(idx, "test")

        # Other base untouched
        other_corrupts = list(tmp_path.glob("other.faiss.corrupt-*"))
        assert len(other_corrupts) == keep + 3
        # Our base pruned
        idx_corrupts = list(tmp_path.glob("idx.faiss.corrupt-*"))
        assert len(idx_corrupts) == keep

    def test_prune_failure_does_not_propagate(self, tmp_path):
        """If unlink raises during the sweep, the quarantine call
        still succeeds — retention is best-effort.
        """
        svc = _make_service()
        mod = _import_module()
        keep = mod._QUARANTINE_KEEP_RECENT

        # Seed enough corrupts to trigger pruning.
        for i in range(keep + 2):
            self._make_quarantined(
                tmp_path, "idx", ns=5000 + i, mtime=5000.0 + i
            )

        before_count = len(list(tmp_path.glob("idx.faiss.corrupt-*")))
        idx, _pkl = self._make_idx(tmp_path)

        # Allow the quarantine rename to succeed, but make unlink raise
        # for the prune sweep. The quarantine itself uses rename, not
        # unlink, so this only affects the retention path.
        real_unlink = Path.unlink

        def selective_unlink(self):
            if ".corrupt-" in self.name:
                raise PermissionError(f"denied: {self}")
            return real_unlink(self)

        with patch.object(Path, "unlink", selective_unlink):
            # Must not raise — sweep is best-effort.
            svc._quarantine_corrupt_index(idx, "test")

        # Quarantine still produced a new pair even though the sweep
        # was unable to delete anything (count went up, not down).
        after_count = len(list(tmp_path.glob("idx.faiss.corrupt-*")))
        assert after_count == before_count + 1, (
            f"expected one new corrupt file; before={before_count}, "
            f"after={after_count}"
        )

    def test_retention_sort_uses_filename_ns_not_mtime(self, tmp_path):
        """On low-resolution filesystems (FAT32, SMB, ext3) mtime can
        round to whole seconds, making same-second quarantines sort
        non-deterministically by ``st_mtime``. The embedded
        ``-<ns>`` suffix is the authoritative ordering — confirm
        retention drops the file with the *lowest* ns even when
        its mtime would mark it as the *newest*.
        """
        svc = _make_service()
        mod = _import_module()
        keep = mod._QUARANTINE_KEEP_RECENT
        import os

        # Seed exactly `keep` corrupts. The OLDEST ns gets the NEWEST
        # mtime; any retention that sorts by mtime would keep it. A
        # sort by ns must prune it once one more quarantine pushes us
        # over the cap.
        ns_values = list(range(6000, 6000 + keep))
        mtimes = list(reversed([7000.0 + i for i in range(keep)]))
        for ns, mtime in zip(ns_values, mtimes):
            self._make_quarantined(tmp_path, "idx", ns=ns, mtime=mtime)

        # Add one more quarantine via the real helper. Its ns is
        # `time.time_ns()` which is far larger than 6000+, so by ns
        # it's the freshest. Force its mtime to ancient so any
        # mtime-based sort would prune *it* instead of the seeded
        # lowest-ns file — that's the regression this test guards.
        idx, _pkl = self._make_idx(tmp_path)
        original_rename = Path.rename

        def rename_then_age(self, target):
            result = original_rename(self, target)
            os.utime(target, (1.0, 1.0))
            return result

        with patch.object(Path, "rename", rename_then_age):
            svc._quarantine_corrupt_index(idx, "test")

        # The lowest-ns seeded file (ns=6000) must be the one pruned.
        # The other seeded files (ns=6001..6004) plus the brand-new
        # one (highest ns) survive.
        assert not (tmp_path / f"idx.faiss.corrupt-{ns_values[0]}").exists()
        for ns in ns_values[1:]:
            assert (tmp_path / f"idx.faiss.corrupt-{ns}").exists(), (
                f"ns={ns} should still exist after retention sort"
            )


class TestFaissWriteLock:
    """Lock infrastructure: identity, parallelism across keys,
    cleanup on user-close.
    """

    def _reset_locks(self):
        mod = _import_module()
        with mod._faiss_write_locks_lock:
            mod._faiss_write_locks.clear()

    def test_same_key_returns_same_lock(self, tmp_path):
        self._reset_locks()
        mod = _import_module()
        p = str(tmp_path / "shared.faiss")
        a = mod._get_faiss_write_lock("u1", p)
        b = mod._get_faiss_write_lock("u1", p)
        assert a is b

    def test_different_key_returns_different_lock(self, tmp_path):
        self._reset_locks()
        mod = _import_module()
        a = mod._get_faiss_write_lock("u1", str(tmp_path / "a.faiss"))
        b = mod._get_faiss_write_lock("u1", str(tmp_path / "b.faiss"))
        c = mod._get_faiss_write_lock("u2", str(tmp_path / "a.faiss"))
        assert a is not b
        assert a is not c

    def test_pop_removes_only_target_user(self, tmp_path):
        self._reset_locks()
        mod = _import_module()
        u1_lock = mod._get_faiss_write_lock("u1", str(tmp_path / "a.faiss"))
        mod._get_faiss_write_lock("u1", str(tmp_path / "b.faiss"))
        u2_lock = mod._get_faiss_write_lock("u2", str(tmp_path / "a.faiss"))

        mod.pop_faiss_locks_for_user("u1")

        # u1 locks gone (new lookup yields a fresh lock object)
        new_u1 = mod._get_faiss_write_lock("u1", str(tmp_path / "a.faiss"))
        assert new_u1 is not u1_lock
        # u2 lock untouched
        same_u2 = mod._get_faiss_write_lock("u2", str(tmp_path / "a.faiss"))
        assert same_u2 is u2_lock

    def test_concurrent_holders_serialised_for_same_key(self, tmp_path):
        """Two threads acquiring ``_get_faiss_write_lock`` for the same
        ``(username, index_path)`` MUST observe non-overlapping critical
        sections. Regression for the #4197 race.
        """
        self._reset_locks()
        mod = _import_module()
        index_path = tmp_path / "idx.faiss"

        in_flight = 0
        max_in_flight = 0
        observed_overlap = False
        lock_for_counters = threading.Lock()
        first_inside = threading.Event()
        first_can_finish = threading.Event()
        first_holder = [True]  # mutable flag to mark which thread enters first

        def worker():
            with mod._get_faiss_write_lock("u1", str(index_path)):
                nonlocal in_flight, max_in_flight, observed_overlap
                with lock_for_counters:
                    in_flight += 1
                    if in_flight > max_in_flight:
                        max_in_flight = in_flight
                    if in_flight > 1:
                        observed_overlap = True
                    am_first = first_holder[0]
                    first_holder[0] = False
                if am_first:
                    first_inside.set()
                    # Hold the critical section open so the second
                    # thread blocks behind us, not races us.
                    first_can_finish.wait(timeout=2.0)
                with lock_for_counters:
                    in_flight -= 1

        t1 = threading.Thread(target=worker)
        t2 = threading.Thread(target=worker)
        t1.start()
        assert first_inside.wait(timeout=2.0)
        t2.start()
        first_can_finish.set()
        t1.join(timeout=3.0)
        t2.join(timeout=3.0)

        assert not observed_overlap, (
            "Two threads held the same FAISS write lock concurrently."
        )
        assert max_in_flight == 1

    def test_different_keys_run_in_parallel(self, tmp_path):
        """Two threads acquiring locks for different
        ``(username, index_path)`` keys must run simultaneously.
        """
        self._reset_locks()
        mod = _import_module()
        barrier = threading.Barrier(2, timeout=2.0)
        idx1 = tmp_path / "a.faiss"
        idx2 = tmp_path / "b.faiss"

        def worker(index_path):
            with mod._get_faiss_write_lock("u1", str(index_path)):
                # If the lock blocked us, barrier.wait would time out
                barrier.wait()

        t1 = threading.Thread(target=worker, args=(idx1,))
        t2 = threading.Thread(target=worker, args=(idx2,))
        t1.start()
        t2.start()
        t1.join(timeout=3.0)
        t2.join(timeout=3.0)
        # If barrier broke (timeout), one of the threads would be alive
        assert not t1.is_alive()
        assert not t2.is_alive()


class TestNoWalCheckpointInIndexingPath:
    """Regression guard: the per-document PRAGMA wal_checkpoint(FULL)
    has been removed (#4197 secondary fix from PR #3539). Re-adding it
    would re-introduce 'database is locked' errors under concurrent
    bulk indexing.
    """

    def test_module_source_has_no_executable_wal_checkpoint(self):
        import inspect

        mod = _import_module()
        src = inspect.getsource(mod)
        # Strip out comment / docstring lines so the regression test
        # tolerates explanatory references like 'PRAGMA wal_checkpoint'
        # in comments, but fails if any code line re-adds the call.
        for lineno, line in enumerate(src.splitlines(), 1):
            stripped = line.strip()
            if not stripped:
                continue
            if stripped.startswith("#"):
                continue
            assert "wal_checkpoint(" not in stripped, (
                f"library_rag_service.py:{lineno} re-introduced "
                f"a wal_checkpoint() call: {line!r}"
            )


class TestMergeAndPersistLocked:
    """Read-modify-write under the FAISS write lock.

    Regression for the AI-review concern on #4200: two workers
    indexing different documents into the same collection used to
    both load on-disk state X, each add their own chunk in memory,
    then save in sequence — last writer wins, the loser's chunks
    were lost from the FAISS file (chunks survived in the DB so a
    rebuild recovered them, but the index file was wrong until then).

    The helper reloads from disk under the lock before adding, so
    the second writer sees the first writer's save and merges
    instead of overwriting.
    """

    def _reset_locks(self):
        mod = _import_module()
        with mod._faiss_write_locks_lock:
            mod._faiss_write_locks.clear()

    def test_reload_picks_up_concurrent_writers_save(self, tmp_path):
        """The merge helper must reload from disk before adding so
        chunks committed by a concurrent writer between the caller's
        in-memory load and this lock acquisition are preserved.
        """
        self._reset_locks()
        mod = _import_module()
        index_path = tmp_path / "idx.faiss"
        # The on-disk file must exist + verify so the helper actually
        # takes the reload branch (not "no on-disk version, keep
        # stale in-memory").
        index_path.touch()

        svc = _make_service()
        svc.embedding_manager = MagicMock()
        svc.integrity_manager = MagicMock()
        svc.integrity_manager.verify_file.return_value = (True, None)
        svc.rag_index_record = MagicMock(id=42)

        # Caller's stale in-memory state knows about no chunks.
        stale_index = MagicMock()
        stale_index.docstore._dict = {}
        svc.faiss_index = stale_index

        # Disk state (loaded by the reload step) already has another
        # writer's chunk "concurrent-id".
        fresh_index = MagicMock()
        fresh_index.docstore._dict = {"concurrent-id": MagicMock()}

        new_chunk = LangchainDocument(page_content="our new content")

        with patch.object(mod, "safe_load_faiss", return_value=fresh_index):
            stats = svc._merge_and_persist_locked(
                index_path,
                [new_chunk],
                ["our-id"],
                force_reindex=False,
            )

        # Helper reloaded → svc.faiss_index is now the fresh state
        assert svc.faiss_index is fresh_index
        # Our chunk was added to the fresh state (not the stale one)
        fresh_index.add_documents.assert_called_once()
        added_ids = fresh_index.add_documents.call_args[1]["ids"]
        assert added_ids == ["our-id"]
        # The stale state was NOT mutated (otherwise we'd be saving it)
        stale_index.add_documents.assert_not_called()
        # save_local was called on the fresh (merged) state
        fresh_index.save_local.assert_called_once()
        assert stats["added"] == 1
        assert stats["added_ids"] == ["our-id"]

    def test_reload_skips_when_existing_id_now_on_disk(self, tmp_path):
        """If the caller wanted to add id X but a concurrent writer
        already saved X to disk, dedup against the fresh state must
        drop X (idempotent re-index).
        """
        self._reset_locks()
        mod = _import_module()
        index_path = tmp_path / "idx.faiss"
        index_path.touch()

        svc = _make_service()
        svc.embedding_manager = MagicMock()
        svc.integrity_manager = MagicMock()
        svc.integrity_manager.verify_file.return_value = (True, None)
        svc.rag_index_record = MagicMock(id=42)
        svc.faiss_index = MagicMock()  # stale, irrelevant

        # Disk state already has the same id we're about to add.
        fresh_index = MagicMock()
        fresh_index.docstore._dict = {"shared-id": MagicMock()}

        with patch.object(mod, "safe_load_faiss", return_value=fresh_index):
            stats = svc._merge_and_persist_locked(
                index_path,
                [LangchainDocument(page_content="duplicate")],
                ["shared-id"],
                force_reindex=False,
            )

        # No add (already on disk); save still happens (idempotent +
        # touches mtime so the integrity record updates).
        fresh_index.add_documents.assert_not_called()
        fresh_index.save_local.assert_called_once()
        assert stats["added"] == 0
        assert stats["skipped"] == 1

    def test_force_reindex_deletes_after_reload(self, tmp_path):
        """force_reindex deletes the matching IDs from the FRESH
        on-disk state (not the caller's stale in-memory state)
        before re-adding.
        """
        self._reset_locks()
        mod = _import_module()
        index_path = tmp_path / "idx.faiss"
        index_path.touch()

        svc = _make_service()
        svc.embedding_manager = MagicMock()
        svc.integrity_manager = MagicMock()
        svc.integrity_manager.verify_file.return_value = (True, None)
        svc.rag_index_record = MagicMock(id=42)
        svc.faiss_index = MagicMock()  # stale

        fresh_index = MagicMock()
        fresh_index.docstore._dict = {"to-update": MagicMock()}

        with patch.object(mod, "safe_load_faiss", return_value=fresh_index):
            svc._merge_and_persist_locked(
                index_path,
                [LangchainDocument(page_content="updated")],
                ["to-update"],
                force_reindex=True,
            )

        # Old copy removed from FRESH state
        fresh_index.delete.assert_called_once_with(["to-update"])
        # New copy added (no dedup under force_reindex)
        fresh_index.add_documents.assert_called_once()
        added_ids = fresh_index.add_documents.call_args[1]["ids"]
        assert added_ids == ["to-update"]

    def test_reload_failure_falls_back_to_in_memory(self, tmp_path):
        """If the loader raises (torn write, partial pickle),
        the merge helper keeps the caller's in-memory state instead
        of losing the write entirely.
        """
        self._reset_locks()
        mod = _import_module()
        index_path = tmp_path / "idx.faiss"
        index_path.touch()

        svc = _make_service()
        svc.embedding_manager = MagicMock()
        svc.integrity_manager = MagicMock()
        svc.integrity_manager.verify_file.return_value = (True, None)
        svc.rag_index_record = MagicMock(id=42)

        in_memory = MagicMock()
        in_memory.docstore._dict = {}
        svc.faiss_index = in_memory

        with patch.object(
            mod, "safe_load_faiss", side_effect=RuntimeError("torn pickle")
        ):
            svc._merge_and_persist_locked(
                index_path,
                [LangchainDocument(page_content="kept")],
                ["kept-id"],
                force_reindex=False,
            )

        # Reload failed → kept the in-memory object, added to it,
        # saved it. The write is preserved.
        assert svc.faiss_index is in_memory
        in_memory.add_documents.assert_called_once()
        in_memory.save_local.assert_called_once()

    def test_concurrent_writers_both_chunks_survive(self, tmp_path):
        """End-to-end regression: two workers each call
        ``_merge_and_persist_locked`` for the same on-disk index.
        Each starts from a stale in-memory snapshot. After both
        complete, the on-disk state must contain BOTH chunks.

        Before the read-modify-write fix, this was last-writer-wins:
        worker A's chunk was overwritten by worker B's save because
        B never reloaded A's write.
        """
        self._reset_locks()
        mod = _import_module()
        index_path = tmp_path / "shared.faiss"
        index_path.touch()

        # Single shared "disk state" as a dict — proxies for what
        # would be on disk via FAISS.load_local/save_local.
        disk_state: dict = {}
        disk_lock = threading.Lock()

        def make_index_proxy():
            """Each call returns a FAISS-like proxy reading/writing
            the shared disk_state. Mimics a fresh load_local()."""
            ix = MagicMock()
            # Snapshot disk_state at load time (proxy for what
            # FAISS.load_local would return).
            snapshot = dict(disk_state)
            ix.docstore = MagicMock()
            ix.docstore._dict = snapshot

            def add(chunks, ids):
                for cid in ids:
                    snapshot[cid] = "doc"

            def save(_folder, index_name=None):
                # Atomically replace disk_state with this index's
                # current snapshot — mirrors save_local's truncating
                # semantics for the test.
                with disk_lock:
                    disk_state.clear()
                    disk_state.update(snapshot)

            ix.add_documents.side_effect = add
            ix.save_local.side_effect = save
            return ix

        def worker(cid: str, start_event: threading.Event):
            svc = _make_service()
            svc.embedding_manager = MagicMock()
            svc.integrity_manager = MagicMock()
            svc.integrity_manager.verify_file.return_value = (True, None)
            svc.rag_index_record = MagicMock(id=99)
            # Pre-merge in-memory state is stale (empty)
            svc.faiss_index = MagicMock()
            svc.faiss_index.docstore = MagicMock()
            svc.faiss_index.docstore._dict = {}
            start_event.wait()  # Race start
            svc._merge_and_persist_locked(
                index_path,
                [LangchainDocument(page_content=f"chunk-{cid}")],
                [cid],
                force_reindex=False,
            )

        go = threading.Event()
        t1 = threading.Thread(target=worker, args=("A", go))
        t2 = threading.Thread(target=worker, args=("B", go))

        # Patch safe_load_faiss ONCE, around both threads. Patching
        # inside each worker (the original shape) is a hidden race:
        # ``patch.object`` rewrites a module attribute and is not
        # thread-safe — when the first worker's
        # ``with patch.object(...)`` block exits, it restores whatever
        # it captured as "the original" at entry, which can be the
        # second worker's lambda OR the real ``safe_load_faiss``
        # (depending on which thread entered patch first). If the real
        # ``safe_load_faiss`` becomes active mid-test, the second worker's
        # reload step raises on the empty touched ``shared.faiss``,
        # the production code's ``except`` branch falls back to the
        # caller's stale in-memory MagicMock, and the worker's save is
        # a no-op (the setup MagicMock has no ``save_local`` side
        # effect). Result: that worker's chunk never reaches
        # ``disk_state``, the assertion fails, and the test wrongly
        # blames a regression in the production lock. Applying the
        # patch around both threads removes the race entirely — the
        # patched value is the only value visible while either worker
        # runs.
        with patch.object(
            mod,
            "safe_load_faiss",
            side_effect=lambda *a, **kw: make_index_proxy(),
        ):
            t1.start()
            t2.start()
            go.set()
            t1.join(timeout=5.0)
            t2.join(timeout=5.0)

        # Both writers' chunks present — read-modify-write under the
        # lock did its job. Without the fix, only one of {A, B} would
        # remain.
        assert "A" in disk_state, (
            f"Worker A's chunk lost; disk_state={disk_state}. "
            f"Read-modify-write race regressed."
        )
        assert "B" in disk_state, (
            f"Worker B's chunk lost; disk_state={disk_state}. "
            f"Read-modify-write race regressed."
        )
