fix(python-sdk): close the per-call httpx client instead of dropping it

packages/python-sdk-memwal/memwal/client.py

@@ -28,6 +28,7 @@
 import asyncio
 import base64
 import json
+import logging
 import random
 import time
 import uuid
@@ -90,6 +91,9 @@
 )
 
 
+logger = logging.getLogger("memwal")
+
+
 # ============================================================
 # Polling helpers (PR #121 parity with TS SDK)
 # ============================================================
@@ -1277,6 +1281,38 @@ def __init__(self, job_id: str, timeout_ms: int) -> None:
         self.timeout_ms = timeout_ms
 
 
+async def _discard_http_client(memwal: MemWal) -> None:
+    """Close the cached httpx client, and drop it either way.
+
+    ``close()`` can fail when the client belongs to an event loop that has
+    already finished — the caller still needs ``_client`` cleared so the next
+    request builds one in a loop that is actually running.
+    """
+    try:
+        await memwal.close()
+    except Exception:
+        logger.debug("Closing the cached HTTP client failed", exc_info=True)
+    finally:
+        memwal._client = None
+
+
+async def _with_fresh_http_client(memwal: MemWal, coro: Any) -> Any:
+    """Run ``coro`` with an httpx client owned by the *current* event loop.
+
+    The sync entry points execute each coroutine in a throwaway ``asyncio.run()``
+    loop, and an ``httpx.AsyncClient`` is bound to the loop that created it, so
+    one can never be reused across calls. Both ends are closed rather than just
+    dropped (GH #606): on the way in for anything an earlier loop left behind —
+    e.g. a caller mixing ``await memwal.recall()`` with the sync wrapper — and in
+    ``finally`` for the client this call created, while its loop is still alive.
+    """
+    await _discard_http_client(memwal)
+    try:
+        return await coro
+    finally:
+        await _discard_http_client(memwal)
+
+
 class MemWalSync:
     """Synchronous wrapper around the async :class:`MemWal` client.
 
@@ -1326,21 +1362,20 @@ def _run(self, coro: Any) -> Any:
         except RuntimeError:
             loop = None
 
-        # Reset the httpx client before every asyncio.run() path so it is
-        # recreated inside the loop that will use it. This matters in
-        # notebooks/Jupyter where the sync wrapper runs coroutines in worker
-        # threads with short-lived event loops.
-        self._inner._client = None
+        # The httpx client is created and closed inside the same short-lived
+        # loop that uses it. This matters in notebooks/Jupyter where the sync
+        # wrapper runs coroutines in worker threads with their own event loops.
+        wrapped = _with_fresh_http_client(self._inner, coro)
 
         if loop is not None and loop.is_running():
             # Already inside an event loop (e.g. Jupyter).
             # Create a new loop in a thread.
             import concurrent.futures
 
             with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
-                return pool.submit(asyncio.run, coro).result()
+                return pool.submit(asyncio.run, wrapped).result()
         else:
-            return asyncio.run(coro)
+            return asyncio.run(wrapped)
 
     def remember(
         self,

packages/python-sdk-memwal/memwal/middleware.py

@@ -54,7 +54,7 @@
     Optional,
 )
 
-from .client import MemWal
+from .client import MemWal, _with_fresh_http_client
 from .types import RecallMemory
 
 if TYPE_CHECKING:
@@ -579,8 +579,7 @@ def _wrap_sync_openai(
 
     def _run_memwal(coro_factory: Callable[[], Any]) -> Any:
         # Keep httpx clients bound to the short-lived loop that uses them.
-        memwal._client = None
-        return _run_blocking(coro_factory)
+        return _run_blocking(lambda: _with_fresh_http_client(memwal, coro_factory()))
 
     def patched_create(*args: Any, **kwargs: Any) -> Any:
         messages = kwargs.get("messages") or (args[0] if args else None)

packages/python-sdk-memwal/tests/test_client.py

@@ -120,9 +120,26 @@ def memwal_client() -> MemWal:
 # ============================================================
 
 
+class _FakeHttpClient:
+    """Stand-in for httpx.AsyncClient, tracking whether it was closed."""
+
+    def __init__(self) -> None:
+        self.is_closed = False
+
+    async def aclose(self) -> None:
+        self.is_closed = True
+
+
 class _SyncRunInner:
+    """Minimal stand-in mirroring MemWal's httpx client lifecycle."""
+
     def __init__(self) -> None:
-        self._client = object()
+        self._client: Any = _FakeHttpClient()
+
+    async def close(self) -> None:
+        if self._client is not None and not self._client.is_closed:
+            await self._client.aclose()
+            self._client = None
 
 
 class TestMemWalSyncRun:
@@ -138,6 +155,57 @@ async def operation() -> str:
         assert result == "ok"
         assert inner._client is None
 
+    async def test_closes_the_client_it_replaces(self) -> None:
+        """GH #606: _run() used to null out _client without closing it, leaking
+        the connection pool of every client an earlier event loop left behind."""
+        inner = _SyncRunInner()
+        orphan = inner._client
+        sync = MemWalSync(inner)  # type: ignore[arg-type]
+
+        async def operation() -> str:
+            return "ok"
+
+        assert not orphan.is_closed
+        assert sync._run(operation()) == "ok"
+
+        assert orphan.is_closed, "the replaced httpx client was never closed"
+        assert inner._client is None
+
+    async def test_closes_the_client_created_during_the_call(self) -> None:
+        """The per-call client is closed inside the loop that created it, so
+        repeated sync calls do not accumulate open pools."""
+        inner = _SyncRunInner()
+        await inner.close()
+        sync = MemWalSync(inner)  # type: ignore[arg-type]
+        created: list = []
+
+        async def operation() -> str:
+            # Stands in for the lazy `_http` property building a client inside
+            # whichever loop is currently running.
+            inner._client = _FakeHttpClient()
+            created.append(inner._client)
+            return "ok"
+
+        assert sync._run(operation()) == "ok"
+
+        assert len(created) == 1
+        assert created[0].is_closed, "the per-call httpx client was left open"
+        assert inner._client is None
+
+    async def test_closes_client_even_when_the_operation_raises(self) -> None:
+        inner = _SyncRunInner()
+        orphan = inner._client
+        sync = MemWalSync(inner)  # type: ignore[arg-type]
+
+        async def failing() -> str:
+            raise ValueError("boom")
+
+        with pytest.raises(ValueError, match="boom"):
+            sync._run(failing())
+
+        assert orphan.is_closed
+        assert inner._client is None
+
 
 # ============================================================
 # remember() tests