7a0da7932b
OSV-Scanner (Scheduled) / scan-scheduled (push) Failing after 0s
Create Release / test-gate (push) Has been cancelled
Create Release / release-gate (push) Has been cancelled
Create Release / ci-gate (push) Has been cancelled
Create Release / version-check (push) Has been cancelled
Create Release / e2e-test-gate (push) Has been cancelled
Create Release / responsive-test-gate (push) Has been cancelled
Create Release / compat-test-gate (push) Has been cancelled
Create Release / compose-integration-gate (push) Has been cancelled
Create Release / vulture-gate (push) Has been cancelled
Create Release / build (push) Has been cancelled
Create Release / provenance (push) Has been cancelled
Create Release / prerelease-docker (push) Has been cancelled
Create Release / publish-docker (push) Has been cancelled
Create Release / create-release (push) Has been cancelled
Create Release / cleanup-changelog (push) Has been cancelled
Create Release / trigger-pypi (push) Has been cancelled
Create Release / monitor-pypi (push) Has been cancelled
Create Release / Clean up orphan prerelease tags and signatures (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-form] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-metrics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-workflow] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-core] (push) Has been cancelled
CodeQL Advanced / Analyze (javascript-typescript) (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [history-news] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [library] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [link-analytics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-core] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-lifecycle] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [error-benchmark] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) (push) Has been cancelled
Docker Tests (Consolidated) / Accessibility Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Unit Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Example Tests (push) Has been cancelled
Docker Tests (Consolidated) / Production Image Smoke Test (push) Has been cancelled
Docker Tests (Consolidated) / Infrastructure Tests (push) Has been cancelled
OSSF Scorecard / OSSF Security Scorecard Analysis (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [mobile] (push) Has been cancelled
Backwards Compatibility / Verify Encryption Constants (push) Has been cancelled
Backwards Compatibility / PyPI Version Compatibility (push) Has been cancelled
Backwards Compatibility / Database Migration Tests (push) Has been cancelled
CodeQL Advanced / Analyze (python) (push) Has been cancelled
Docker Tests (Consolidated) / detect-changes (push) Has been cancelled
Docker Tests (Consolidated) / Build Test Image (push) Has been cancelled
Docker Tests (Consolidated) / All Pytest Tests + Coverage (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [accessibility] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [api-crud] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-login] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-register] (push) Has been cancelled
855 lines
31 KiB
Python
855 lines
31 KiB
Python
"""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"
|
||
)
|