"""Tests for ``_close_base_llm`` covering both sync and async httpx clients.

Background (issue #3816): ``ChatOllama`` owns both ``_client`` (sync
``ollama.Client`` wrapping ``httpx.Client``) and ``_async_client``
(async ``ollama.AsyncClient`` wrapping ``httpx.AsyncClient``). Earlier
versions of ``_close_base_llm`` only closed the sync side, leaking the
async transport per ``ainvoke()`` call — visible as ``a_inode [eventpoll]``
FDs in the issue's lsof dump.
"""

import asyncio
import gc
import os
import resource
import sys
from unittest.mock import Mock

import httpx
import pytest

from local_deep_research.utilities.llm_utils import _close_base_llm


def _open_fd_count() -> int:
    """Test-local file-descriptor counter.

    Inlined here to avoid coupling these tests to a private helper in
    an unrelated production module. On Linux uses ``/proc/self/fd``
    (fast); on macOS falls back to scanning ``RLIMIT_NOFILE``.
    """
    try:
        return len(os.listdir("/proc/self/fd"))
    except (FileNotFoundError, OSError):
        soft_limit = resource.getrlimit(resource.RLIMIT_NOFILE)[0]
        count = 0
        for fd in range(soft_limit):
            try:
                os.fstat(fd)
                count += 1
            except OSError:
                pass
        return count


def _make_fake_chat_ollama(*, sync_close=None, async_aclose=None):
    """Build a stand-in for ``ChatOllama`` with the same private-attr shape.

    The module-string check in ``_close_base_llm`` looks at
    ``type(ollama_client).__module__`` — we set ``__module__`` on the mock's
    type to "ollama" so the introspection treats them as real ollama clients.
    """
    llm = Mock(spec=[])  # spec=[] means hasattr returns False for everything

    if sync_close is not None:
        sync_httpx = Mock(spec=["close"])
        sync_httpx.close = sync_close
        sync_ollama = type("FakeSyncOllama", (), {})()
        sync_ollama._client = sync_httpx
        type(sync_ollama).__module__ = "ollama"
        llm._client = sync_ollama
    else:
        llm._client = None

    if async_aclose is not None:
        async_httpx = Mock(spec=["aclose"])
        async_httpx.aclose = async_aclose
        async_ollama = type("FakeAsyncOllama", (), {})()
        async_ollama._client = async_httpx
        type(async_ollama).__module__ = "ollama"
        llm._async_client = async_ollama
    else:
        llm._async_client = None

    return llm


class TestCloseBaseLLMSync:
    """The sync httpx client must be closed on every call until idempotent."""

    def test_closes_sync_httpx_client(self):
        sync_close = Mock()
        llm = _make_fake_chat_ollama(sync_close=sync_close)

        _close_base_llm(llm)

        sync_close.assert_called_once()

    def test_idempotent_via_ldr_closed_flag(self):
        sync_close = Mock()
        llm = _make_fake_chat_ollama(sync_close=sync_close)

        _close_base_llm(llm)
        _close_base_llm(llm)
        _close_base_llm(llm)

        # Despite three calls, close runs once — _ldr_closed sentinel guards.
        sync_close.assert_called_once()

    def test_swallows_close_exception_and_still_marks_closed(self):
        sync_close = Mock(side_effect=RuntimeError("boom"))
        llm = _make_fake_chat_ollama(sync_close=sync_close)

        # Must not propagate; logs at warning.
        _close_base_llm(llm)

        sync_close.assert_called_once()
        # Subsequent call is skipped by _ldr_closed (no infinite retry).
        _close_base_llm(llm)
        sync_close.assert_called_once()

    def test_handles_missing_async_client_gracefully(self):
        # No _async_client attr at all — must not crash.
        sync_close = Mock()
        llm = _make_fake_chat_ollama(sync_close=sync_close)
        del llm._async_client  # simulate older ollama versions

        _close_base_llm(llm)

        sync_close.assert_called_once()


class TestCloseBaseLLMAsync:
    """The async httpx client must be closed via ``asyncio.run`` when no loop
    is running, and via a brief daemon thread when one is."""

    def test_closes_async_httpx_client_via_asyncio_run(self):
        called = {"count": 0}

        async def fake_aclose():
            called["count"] += 1

        llm = _make_fake_chat_ollama(async_aclose=fake_aclose)

        # No running loop here — _close_base_llm should spin one via
        # asyncio.run() and await aclose().
        _close_base_llm(llm)

        assert called["count"] == 1

    def test_async_close_is_idempotent(self):
        called = {"count": 0}

        async def fake_aclose():
            called["count"] += 1

        llm = _make_fake_chat_ollama(async_aclose=fake_aclose)

        _close_base_llm(llm)
        _close_base_llm(llm)

        # _ldr_closed sentinel prevents the second aclose.
        assert called["count"] == 1

    def test_closes_async_inside_running_loop_via_thread(self):
        """Regression for the v1.6.10 leak. ``_close_base_llm`` used to
        skip the async close when invoked inside a running asyncio loop
        and rely on a non-existent "loop owner" cleanup — so the inner
        ``httpx.AsyncClient`` (and its ``epoll_create`` FD) was silently
        abandoned. The current implementation must run the close in a
        brief daemon thread whose own ``asyncio.run`` is independent of
        the caller's loop.
        """
        called = {"count": 0}

        async def fake_aclose():
            called["count"] += 1

        llm = _make_fake_chat_ollama(async_aclose=fake_aclose)

        async def driver():
            _close_base_llm(llm)

        asyncio.run(driver())

        # aclose ran exactly once (via the cleanup thread, not skipped).
        assert called["count"] == 1
        # And _ldr_closed IS set on success — subsequent calls short-circuit.
        async_httpx = llm._async_client._client
        assert async_httpx._ldr_closed is True

    def test_in_loop_close_is_idempotent(self):
        """A close fired from inside a loop should set ``_ldr_closed`` just
        like the no-loop path, so repeat calls don't re-spawn the cleanup
        thread or re-run ``aclose``."""
        called = {"count": 0}

        async def fake_aclose():
            called["count"] += 1

        llm = _make_fake_chat_ollama(async_aclose=fake_aclose)

        async def driver():
            _close_base_llm(llm)
            _close_base_llm(llm)
            _close_base_llm(llm)

        asyncio.run(driver())

        assert called["count"] == 1

    def test_in_loop_close_timeout_does_not_mark_closed(self):
        """If the cleanup thread is still alive after the 5-second join
        (e.g. ``aclose`` is blocked on a stuck server), the sentinel must
        NOT be set so a later call can retry — and the FD leak is at
        least visible via WARNING log instead of silent drift.
        """
        import threading
        from unittest.mock import patch

        release = threading.Event()
        aclose_started = threading.Event()

        async def slow_aclose():
            aclose_started.set()
            # Block until released (or the test's shortened join fires).
            await asyncio.get_event_loop().run_in_executor(
                None, release.wait, 30
            )

        llm = _make_fake_chat_ollama(async_aclose=slow_aclose)

        original_thread = threading.Thread

        class _ShortJoinThread(original_thread):
            def join(self, timeout=None):
                # Tighten the production 5s wait to 200ms for the test
                # so we don't actually sit here for 5 seconds.
                return super().join(timeout=0.2)

        try:
            with patch("threading.Thread", _ShortJoinThread):

                async def driver():
                    _close_base_llm(llm)

                asyncio.run(driver())
        finally:
            release.set()

        # ``aclose`` started but the join timed out before it could
        # finish — sentinel must be unset so the FD is not silently
        # leaked (a subsequent _close_base_llm call should retry).
        assert aclose_started.is_set()
        async_httpx = llm._async_client._client
        assert not getattr(async_httpx, "_ldr_closed", False)

    def test_swallows_async_close_exception(self):
        async def fake_aclose():
            raise RuntimeError("boom")

        llm = _make_fake_chat_ollama(async_aclose=fake_aclose)

        # Must not propagate; logs at warning. _ldr_closed is set so we don't
        # retry endlessly on a known-broken close.
        _close_base_llm(llm)

        async_httpx = llm._async_client._client
        assert getattr(async_httpx, "_ldr_closed", False) is True


class TestCloseBaseLLMBoth:
    """Sync and async sides should both close in the common case."""

    def test_closes_both_sync_and_async(self):
        sync_close = Mock()
        async_called = {"count": 0}

        async def fake_aclose():
            async_called["count"] += 1

        llm = _make_fake_chat_ollama(
            sync_close=sync_close, async_aclose=fake_aclose
        )

        _close_base_llm(llm)

        sync_close.assert_called_once()
        assert async_called["count"] == 1


class TestCloseBaseLLMNonOllama:
    """Non-Ollama LLMs must be left alone. ChatAnthropic/ChatOpenAI use
    @lru_cache'd shared httpx clients that must NOT be closed."""

    def test_skips_non_ollama_module(self):
        llm = Mock(spec=[])
        non_ollama = type("OpenAIClient", (), {})()
        non_ollama._client = Mock()
        type(non_ollama).__module__ = "openai"  # not "ollama"
        llm._client = non_ollama
        llm._async_client = None

        _close_base_llm(llm)

        non_ollama._client.close.assert_not_called()

    def test_delegates_to_wrapper_close_method(self):
        # If the LLM type defines close(), delegate to that and skip
        # introspection. (Wrappers like ProcessingLLMWrapper take this path.)
        class FakeWrapper:
            def __init__(self):
                self.closed = False

            def close(self):
                self.closed = True

        wrapper = FakeWrapper()
        _close_base_llm(wrapper)

        assert wrapper.closed is True


class TestCloseBaseLLMRealHttpxAsync:
    """Empirical validation against a real ``httpx.AsyncClient`` — covers the
    actual #3816 leak shape: a client created inside one ``asyncio.run``
    (loop A) survives loop A's close and must be released by
    ``_close_base_llm`` spinning a fresh loop B. No Ollama server required.
    """

    @pytest.mark.skipif(
        sys.platform == "win32",
        reason="Linux/macOS-specific FD semantics",
    )
    def test_real_async_client_created_in_closed_loop_is_closed(self):
        async def _make():
            return httpx.AsyncClient()

        async_httpx = asyncio.run(_make())  # loop A constructs, then closes
        assert async_httpx.is_closed is False  # client survived loop A

        async_ollama = type("FakeAsyncOllama", (), {})()
        async_ollama._client = async_httpx
        type(async_ollama).__module__ = "ollama"
        llm = Mock(spec=[])
        llm._client = None
        llm._async_client = async_ollama

        _close_base_llm(llm)

        assert async_httpx.is_closed is True
        assert getattr(async_httpx, "_ldr_closed", False) is True

    @pytest.mark.skipif(
        sys.platform == "win32",
        reason="Linux/macOS-specific FD semantics",
    )
    def test_real_async_client_close_is_idempotent(self):
        async def _make():
            return httpx.AsyncClient()

        async_httpx = asyncio.run(_make())
        async_ollama = type("FakeAsyncOllama", (), {})()
        async_ollama._client = async_httpx
        type(async_ollama).__module__ = "ollama"
        llm = Mock(spec=[])
        llm._client = None
        llm._async_client = async_ollama

        _close_base_llm(llm)
        _close_base_llm(llm)  # sentinel short-circuits; must not raise

        assert async_httpx.is_closed is True

    @pytest.mark.fd_canary
    @pytest.mark.timeout(180)
    @pytest.mark.skipif(
        sys.platform == "win32",
        reason="Linux/macOS-specific FD semantics",
    )
    def test_no_fd_growth_across_repeated_close_cycles(self):
        # Unit-level analogue of the PR's manual `lsof | grep -c eventpoll`
        # smoke. A real per-cycle leak (~1 FD/iter as observed in #3816)
        # would push the delta well past the +8 slack.
        #
        # Sizing rationale (#4214 + #3816 follow-up):
        # Each cycle invokes ``asyncio.run`` twice — once to construct the
        # client in a (then-closed) loop A, once inside ``_close_base_llm``
        # to spin loop B for ``aclose()``. On Py 3.14 these transient loops
        # leave ambient FDs from selector/self-pipe lazy release. CI on the
        # 5-iter version of this test showed a deterministic +3 delta (58
        # of 58 recent failures), i.e. ~0.6 FDs/cycle steady-state — not
        # absorbable by a single warmup since it's per-cycle, not init-only.
        # At 20 iterations the expected ceiling is ~0.6 × 20 = ~12 in the
        # worst case (observed ~3–4 locally); the +8 slack covers the
        # typical range without masking a real per-iteration leak, which
        # would land at ≥+20 (1 FD/iter, #3816 shape) + ambient → far
        # above the threshold.
        async def _make():
            return httpx.AsyncClient()

        def _one_cycle():
            async_httpx = asyncio.run(_make())
            async_ollama = type("FakeAsyncOllama", (), {})()
            async_ollama._client = async_httpx
            type(async_ollama).__module__ = "ollama"
            llm = Mock(spec=[])
            llm._client = None
            llm._async_client = async_ollama

            _close_base_llm(llm)

            del llm, async_ollama, async_httpx

        # Warmup: one cycle before measuring absorbs one-time init drift
        # (lazy imports, logging handlers, asyncio internals first-init).
        _one_cycle()
        gc.collect()

        before = _open_fd_count()

        for _ in range(20):
            _one_cycle()
            gc.collect()

        gc.collect()
        after = _open_fd_count()

        assert after - before <= 8, (
            f"FD count climbed across close cycles: "
            f"before={before}, after={after}"
        )

    @pytest.mark.fd_canary
    @pytest.mark.timeout(120)
    @pytest.mark.skipif(
        sys.platform == "win32",
        reason="Linux/macOS-specific FD semantics",
    )
    def test_no_fd_growth_when_closed_inside_running_loop(self):
        """The user-facing regression. Before the fix, calling
        ``_close_base_llm`` inside a running loop silently skipped the
        async close — every iteration leaked ~1 ``epoll_create`` FD.
        After the fix, the cleanup thread closes the client even when
        a loop is running, so the FD count stays flat across iterations.
        """

        async def _close_from_inside_loop():
            async_httpx = httpx.AsyncClient()
            async_ollama = type("FakeAsyncOllama", (), {})()
            async_ollama._client = async_httpx
            type(async_ollama).__module__ = "ollama"
            llm = Mock(spec=[])
            llm._client = None
            llm._async_client = async_ollama

            # _close_base_llm is invoked while this loop is running —
            # the exact scenario that leaked before the fix.
            _close_base_llm(llm)

            assert async_httpx.is_closed is True

        # Warmup cycle to absorb one-time init drift (cleanup-thread
        # startup, lazy imports) — see the sibling repeated-close test.
        asyncio.run(_close_from_inside_loop())
        gc.collect()

        before = _open_fd_count()

        for _ in range(5):
            asyncio.run(_close_from_inside_loop())
            gc.collect()

        gc.collect()
        after = _open_fd_count()

        assert after - before <= 2, (
            f"FD count climbed across in-loop close cycles "
            f"(this is the #3816-shaped leak): "
            f"before={before}, after={after}"
        )


class TestCloseBaseLLMRealLangchain:
    """End-to-end regression coverage against the real langchain LLM classes
    users actually instantiate. Guards against:

    - A future ``langchain_ollama`` version reshaping ``_client`` /
      ``_async_client`` so the introspection silently no-ops and the
      eventpoll-FD leak from #3816 returns.
    - The introspection accidentally tripping on a non-Ollama LLM and
      closing a shared, lru_cache'd httpx client (which would brick the
      module-global pool for all other instances).

    Construction is no-network: ``ChatOllama(host=…)`` and
    ``ChatOpenAI(api_key=…)`` are both inert until first ``invoke`` /
    ``ainvoke``.
    """

    def test_real_chatollama_through_wrapper_closes_both_clients(self):
        from langchain_ollama import ChatOllama

        from local_deep_research.config.llm_config import (
            wrap_llm_without_think_tags,
        )
        from local_deep_research.utilities.resource_utils import safe_close

        llm = ChatOllama(model="test", base_url="http://localhost:1")
        sync_httpx = llm._client._client
        async_httpx = llm._async_client._client
        assert sync_httpx.is_closed is False
        assert async_httpx.is_closed is False

        wrapper = wrap_llm_without_think_tags(llm, settings_snapshot={})

        safe_close(wrapper, "regression test ChatOllama")

        # Both inner httpx clients must be closed by the chain
        # (safe_close → ProcessingLLMWrapper.close → _close_base_llm →
        # sync close + asyncio.run(aclose)). This is the load-bearing
        # assertion for #3816.
        assert sync_httpx.is_closed is True
        assert async_httpx.is_closed is True

    def test_real_chatopenai_through_wrapper_does_not_close_shared_client(
        self,
    ):
        from langchain_openai import ChatOpenAI

        from local_deep_research.config.llm_config import (
            wrap_llm_without_think_tags,
        )
        from local_deep_research.utilities.resource_utils import safe_close

        llm = ChatOpenAI(model="gpt-4", api_key="sk-test")

        # ChatOpenAI's underlying httpx clients live behind the openai SDK
        # (langchain_openai's lru_cache'd ``_SyncHttpxClientWrapper`` /
        # ``_AsyncHttpxClientWrapper``). They are shared across every
        # ChatOpenAI instance — closing one would brick all the others.
        sync_httpx = llm.root_client._client
        async_httpx = llm.root_async_client._client
        assert sync_httpx.is_closed is False
        assert async_httpx.is_closed is False

        wrapper = wrap_llm_without_think_tags(llm, settings_snapshot={})

        safe_close(wrapper, "regression test ChatOpenAI")

        # ChatOpenAI exposes neither ``_client`` nor ``_async_client``;
        # `_close_base_llm`'s introspection short-circuits cleanly and
        # the shared cached httpx clients must remain open.
        assert sync_httpx.is_closed is False
        assert async_httpx.is_closed is False

    def test_full_wrapper_stack_via_rate_limited_closes_both_clients(self):
        """Regression: when rate limiting is enabled, the production
        wrapper stack is ``ProcessingLLMWrapper(RateLimitedLLMWrapper(
        ChatOllama))`` and ``safe_close`` has to recurse through *both*
        wrapper layers to reach ``_close_base_llm``'s introspection.

        The other ``Real Langchain`` tests only exercise the
        ``ProcessingLLMWrapper`` layer. This one specifically guards
        the ``RateLimitedLLMWrapper.close → _close_base_llm`` hop so a
        future change to that wrapper's ``close()`` doesn't silently
        break the chain and let the #3816 leak come back.
        """
        from langchain_ollama import ChatOllama

        from local_deep_research.config.llm_config import (
            wrap_llm_without_think_tags,
        )
        from local_deep_research.utilities.resource_utils import safe_close
        from local_deep_research.web_search_engines.rate_limiting.llm import (
            create_rate_limited_llm_wrapper,
        )

        llm = ChatOllama(model="test", base_url="http://localhost:1")
        sync_httpx = llm._client._client
        async_httpx = llm._async_client._client
        assert sync_httpx.is_closed is False
        assert async_httpx.is_closed is False

        # Build the stack manually — wrap_llm_without_think_tags only
        # adds the rate-limited layer when ``rate_limiting.llm_enabled``
        # is True in the settings, and we want this test to hold
        # regardless of that setting.
        rate_limited = create_rate_limited_llm_wrapper(llm, provider="ollama")
        full = wrap_llm_without_think_tags(rate_limited, settings_snapshot={})

        safe_close(full, "regression test full wrapper stack")

        # Recursion: safe_close(Processing) → Processing.close →
        # _close_base_llm(RateLimited) → hasattr(type, "close") fires →
        # RateLimited.close() → _close_base_llm(ChatOllama) →
        # introspection closes both inner httpx clients.
        assert sync_httpx.is_closed is True
        assert async_httpx.is_closed is True


class TestCloseBaseLLMRealOllamaEmbeddings:
    """End-to-end regression coverage for ``OllamaEmbeddings``.

    After the langchain_community → langchain_ollama migration
    (#4352/#4353), ``OllamaEmbeddings`` carries the same
    ``_client`` / ``_async_client`` shape as ``ChatOllama`` — eagerly
    constructed by a Pydantic ``@model_validator(mode="after")``, with no
    ``close()`` / ``aclose()`` / ``__del__`` of its own. The deprecated
    ``langchain_community.embeddings.OllamaEmbeddings`` was FD-safe by
    accident (``requests.post()`` per call, no persistent client); the
    new class is not. The resource-cleanup doc predicted exactly this
    leak shape; this class is the canary that catches a recurrence if
    a future migration breaks the close path again.

    Construction is no-network: ``OllamaEmbeddings(base_url=…)`` is inert
    until the first ``embed_query`` / ``embed_documents`` call.
    """

    def test_real_ollama_embeddings_closes_both_clients(self):
        from langchain_ollama import OllamaEmbeddings

        embeddings = OllamaEmbeddings(
            model="test", base_url="http://localhost:1"
        )
        sync_httpx = embeddings._client._client
        async_httpx = embeddings._async_client._client
        assert sync_httpx.is_closed is False
        assert async_httpx.is_closed is False

        _close_base_llm(embeddings)

        # Both inner httpx clients must be closed. Sync side is closed
        # synchronously; async side via the ``asyncio.run`` branch (no
        # running loop in this thread).
        assert sync_httpx.is_closed is True
        assert async_httpx.is_closed is True

    def test_real_ollama_embeddings_close_is_idempotent(self):
        from langchain_ollama import OllamaEmbeddings

        embeddings = OllamaEmbeddings(
            model="test", base_url="http://localhost:1"
        )
        async_httpx = embeddings._async_client._client

        _close_base_llm(embeddings)
        _close_base_llm(embeddings)  # sentinel short-circuits; must not raise

        assert async_httpx.is_closed is True
        assert getattr(async_httpx, "_ldr_closed", False) is True

    def test_local_embedding_manager_close_closes_ollama_embeddings(self):
        """Integration: ``LocalEmbeddingManager.close()`` must flow through
        to ``_close_base_llm`` for an Ollama-backed embeddings instance.

        Without this, the ``LibraryRAGService`` close path — which
        cascades to the manager's ``close()`` — leaks the underlying
        httpx clients per RAG request (the #3816-shaped FD ramp, now on
        the embeddings side).
        """
        from langchain_ollama import OllamaEmbeddings

        from local_deep_research.web_search_engines.engines.local_embedding_manager import (
            LocalEmbeddingManager,
        )

        mgr = LocalEmbeddingManager(
            embedding_model="test",
            embedding_model_type="ollama",
            ollama_base_url="http://localhost:1",
            settings_snapshot={},
        )
        # Inject a real embeddings instance directly to bypass the lazy
        # ``embeddings`` property — its provider lookup would touch the
        # settings / database layers, which aren't relevant to this test.
        mgr._embeddings = OllamaEmbeddings(
            model="test", base_url="http://localhost:1"
        )
        sync_httpx = mgr._embeddings._client._client
        async_httpx = mgr._embeddings._async_client._client
        assert sync_httpx.is_closed is False
        assert async_httpx.is_closed is False

        mgr.close()

        assert sync_httpx.is_closed is True
        assert async_httpx.is_closed is True

    def test_local_embedding_manager_close_is_safe_for_non_ollama(self):
        """The close path must not blow up for non-Ollama providers. The
        module-prefix check inside ``_close_base_llm`` ensures the
        ``HuggingFaceEmbeddings`` fallback (and any other non-Ollama
        provider) is a no-op rather than an AttributeError.
        """
        from local_deep_research.web_search_engines.engines.local_embedding_manager import (
            LocalEmbeddingManager,
        )

        mgr = LocalEmbeddingManager(
            embedding_model="test",
            embedding_model_type="sentence_transformers",
            settings_snapshot={},
        )
        # Stand-in with neither ``_client`` nor ``_async_client``.
        non_ollama = type("FakeHFEmbeddings", (), {})()
        mgr._embeddings = non_ollama

        # Must complete cleanly; the introspection short-circuits when
        # the attributes are absent. ``hasattr(type(non_ollama), "close")``
        # is False for our minimal type, so the wrapper-delegation path
        # is also skipped — exactly the behaviour we want.
        mgr.close()

        assert mgr._embeddings is None
        assert mgr._closed is True


class TestLibraryRAGServiceCloseOwnership:
    """``LibraryRAGService`` may be constructed with a caller-supplied
    ``embedding_manager`` (test fixtures, multi-service callers reusing
    one manager). Closing the service must NOT close a manager it didn't
    create — that would burn the caller's reference and surface as a
    use-after-close error in the next call site.
    """

    def test_close_does_not_touch_caller_supplied_manager(self):
        from unittest.mock import Mock

        from local_deep_research.research_library.services.library_rag_service import (
            LibraryRAGService,
        )

        external_manager = Mock(spec=["close"])

        # Bypass the heavy ``__init__`` (DB session, FileIntegrityManager,
        # text splitter) — only the ownership + close branch matters here.
        svc = LibraryRAGService.__new__(LibraryRAGService)
        svc._closed = False
        svc.embedding_manager = external_manager
        svc._owns_embedding_manager = False
        svc.faiss_index = None
        svc.rag_index_record = None
        svc.integrity_manager = None
        svc.text_splitter = None

        svc.close()

        external_manager.close.assert_not_called()
        assert svc.embedding_manager is None

    def test_close_tears_down_owned_manager(self):
        from unittest.mock import Mock

        from local_deep_research.research_library.services.library_rag_service import (
            LibraryRAGService,
        )

        owned_manager = Mock(spec=["close"])

        svc = LibraryRAGService.__new__(LibraryRAGService)
        svc._closed = False
        svc.embedding_manager = owned_manager
        svc._owns_embedding_manager = True
        svc.faiss_index = None
        svc.rag_index_record = None
        svc.integrity_manager = None
        svc.text_splitter = None

        svc.close()

        owned_manager.close.assert_called_once()
        assert svc.embedding_manager is None


class TestInLoopCleanupThreadContract:
    """The in-running-loop branch of ``_close_base_llm`` hands the
    ``aclose()`` off to a daemon thread (see the
    ``else: # A loop is running in this thread`` block that spawns
    a ``ldr-async-llm-close`` thread). Two properties of that thread
    are load-bearing but not pinned by the other tests in this file:

    1. ``daemon=True`` — without this, a cleanup thread blocked on a
       stuck server would prevent Python interpreter shutdown.
    2. The thread is given a recognizable name so operators chasing FD
       leaks can correlate ``threading.enumerate()`` / ``top`` output
       against the leak source.

    Plus the third gap: when the cleanup thread completes BUT its inner
    ``asyncio.run(aclose())`` raised, the main thread still sets
    ``_ldr_closed = True`` in the ``else`` branch reached when
    ``t.is_alive()`` is False. The pre-existing
    ``test_swallows_async_close_exception`` covers this for the no-loop
    branch; this class covers the same invariant for the in-loop branch.
    """

    def test_cleanup_thread_is_daemon_so_shutdown_is_not_blocked(self):
        """Capture the thread object created by the running-loop branch
        and assert ``daemon`` is True. A non-daemon cleanup thread
        blocked on a slow server would hang interpreter shutdown until
        the OS kills the process — exactly the failure mode the
        docstring of ``_close_base_llm`` warns about when motivating
        the brief daemon thread.
        """
        import threading
        from unittest.mock import patch

        async def fake_aclose():
            return None

        llm = _make_fake_chat_ollama(async_aclose=fake_aclose)

        captured = {}
        original_thread = threading.Thread

        class _CapturingThread(original_thread):
            def __init__(self, *args, **kwargs):
                captured["daemon"] = kwargs.get("daemon")
                captured["name"] = kwargs.get("name")
                super().__init__(*args, **kwargs)

        async def driver():
            with patch("threading.Thread", _CapturingThread):
                _close_base_llm(llm)

        asyncio.run(driver())

        assert captured["daemon"] is True, (
            "cleanup thread must be a daemon — otherwise a stuck "
            "aclose() would block Python interpreter shutdown"
        )
        # Name is operator-facing for FD-leak triage; pin a stable
        # prefix rather than the exact string so future tweaks can
        # extend without breaking the test.
        assert captured["name"] is not None
        assert "ldr" in captured["name"].lower()

    def test_in_loop_close_marks_closed_even_when_inner_aclose_raises(
        self,
    ):
        """The cleanup thread's ``_close_in_thread`` runs
        ``asyncio.run(aclose())`` inside a ``try/except Exception`` — if
        ``aclose`` raises, the thread logs a warning and exits cleanly.
        The main thread sees ``t.is_alive() == False`` and falls into
        the ``else`` branch that sets ``_ldr_closed = True``. Pin that
        invariant: the sentinel is the signal to never retry a
        known-broken close, regardless of the branch (no-loop or
        in-loop) that performed it.
        """

        async def boom_aclose():
            raise RuntimeError("simulated upstream failure")

        llm = _make_fake_chat_ollama(async_aclose=boom_aclose)

        async def driver():
            _close_base_llm(llm)

        asyncio.run(driver())

        async_httpx = llm._async_client._client
        assert async_httpx._ldr_closed is True, (
            "Even when aclose raises inside the cleanup thread, the "
            "main thread must mark the client closed to prevent "
            "indefinite retries on a known-broken close"
        )