Key async pool by event loop with WeakKeyDictionary

Address review feedback: async shared pool is now keyed by the running
event loop using weakref.WeakKeyDictionary, so clients in different
asyncio.run() calls get separate pools. close() properly awaits
aclose() on the last release. __del__ uses a sync-only release path.

Async tests converted to async def so they run inside an event loop
and can exercise per-loop pool sharing correctly.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: jrvb-rl

src/runloop_api_client/_base_client.py

@@ -8,6 +8,7 @@
 import asyncio
 import inspect
 import logging
+import weakref
 import platform
 import warnings
 import threading
@@ -93,11 +94,22 @@
 
 # Shared HTTP connection pool state. One pool per client type (sync/async),
 # refcounted so the pool is closed when the last SDK client releases it.
+# The async pool is keyed by event loop because httpx.AsyncClient binds to
+# the loop that created it and cannot be reused across asyncio.run() calls.
 _pool_lock = threading.Lock()
 _shared_sync_client: httpx.Client | None = None
 _shared_sync_refcount: int = 0
-_shared_async_client: httpx.AsyncClient | None = None
-_shared_async_refcount: int = 0
+
+
+class _AsyncPoolState:
+    __slots__ = ("client", "refcount")
+
+    def __init__(self, client: httpx.AsyncClient) -> None:
+        self.client = client
+        self.refcount = 1
+
+
+_async_pools: weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, _AsyncPoolState] = weakref.WeakKeyDictionary()
 
 # TODO: make base page type vars covariant
 SyncPageT = TypeVar("SyncPageT", bound="BaseSyncPage[Any]")
@@ -1516,6 +1528,7 @@ class AsyncAPIClient(BaseClient[httpx.AsyncClient, AsyncStream[Any]]):
     _default_stream_cls: type[AsyncStream[Any]] | None = None
     _uses_shared_pool: bool
     _closed: bool
+    _pool_loop: asyncio.AbstractEventLoop | None
 
     def __init__(
         self,
@@ -1560,29 +1573,39 @@ def __init__(
         )
 
         self._closed = False
+        self._pool_loop = None
 
         if http_client is not None:
             self._client = http_client
             self._uses_shared_pool = False
         elif shared_http_pool:
-            global _shared_async_client, _shared_async_refcount
-            with _pool_lock:
-                if _shared_async_client is None or _shared_async_client.is_closed:
-                    _shared_async_client = AsyncHttpxClientWrapper(
-                        base_url=base_url,
-                        timeout=cast(Timeout, timeout),
-                    )
-                    _shared_async_refcount = 0
-                _shared_async_refcount += 1
-                self._client = _shared_async_client
-            self._uses_shared_pool = True
+            self._acquire_shared_pool(base_url, cast(Timeout, timeout))
         else:
             self._client = AsyncHttpxClientWrapper(
                 base_url=base_url,
                 timeout=cast(Timeout, timeout),
             )
             self._uses_shared_pool = False
 
+    def _acquire_shared_pool(self, base_url: str | URL, timeout: Timeout) -> None:
+        try:
+            loop = asyncio.get_running_loop()
+        except RuntimeError:
+            loop = None
+        with _pool_lock:
+            pool_state = _async_pools.get(loop) if loop is not None else None
+            if pool_state is None or pool_state.client.is_closed:
+                pool_state = _AsyncPoolState(
+                    AsyncHttpxClientWrapper(base_url=base_url, timeout=timeout),
+                )
+                if loop is not None:
+                    _async_pools[loop] = pool_state
+            else:
+                pool_state.refcount += 1
+            self._client = pool_state.client
+        self._pool_loop = loop
+        self._uses_shared_pool = True
+
     def is_closed(self) -> bool:
         return self._closed or self._client.is_closed
 
@@ -1595,36 +1618,59 @@ async def close(self) -> None:
             return
         self._closed = True
         if self._uses_shared_pool:
-            self._release_shared_pool()
+            await self._release_shared_pool()
         else:
             await self._client.aclose()
 
-    def _release_shared_pool(self) -> None:
-        global _shared_async_client, _shared_async_refcount
-        should_close = False
-        client = None
+    async def _release_shared_pool(self) -> None:
+        client: httpx.AsyncClient | None = None
         with _pool_lock:
             if not self._uses_shared_pool:
                 return
             self._uses_shared_pool = False
-            _shared_async_refcount -= 1
-            if _shared_async_refcount <= 0:
-                client = _shared_async_client
-                _shared_async_client = None
-                _shared_async_refcount = 0
-                should_close = True
-        if should_close and client is not None:
+            loop = self._pool_loop
+            if loop is None:
+                return
+            pool = _async_pools.get(loop)
+            if pool is not None:
+                pool.refcount -= 1
+                if pool.refcount <= 0:
+                    client = pool.client
+                    del _async_pools[loop]
+        if client is not None:
             try:
-                loop = asyncio.get_running_loop()
-                loop.create_task(client.aclose())
+                await client.aclose()
+            except Exception:
+                pass
+
+    def _release_shared_pool_sync(self) -> None:
+        """Best-effort synchronous release for use in __del__."""
+        client: httpx.AsyncClient | None = None
+        with _pool_lock:
+            if not self._uses_shared_pool:
+                return
+            self._uses_shared_pool = False
+            loop = self._pool_loop
+            if loop is None:
+                return
+            pool = _async_pools.get(loop)
+            if pool is not None:
+                pool.refcount -= 1
+                if pool.refcount <= 0:
+                    client = pool.client
+                    del _async_pools[loop]
+        if client is not None:
+            try:
+                running_loop = asyncio.get_running_loop()
+                running_loop.create_task(client.aclose())
             except Exception:
                 pass
 
     def __del__(self) -> None:
         if not getattr(self, "_uses_shared_pool", False):
             return
         try:
-            self._release_shared_pool()
+            self._release_shared_pool_sync()
         except Exception:
             pass
 

tests/test_shared_pool.py

@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 import os
+import asyncio
 from typing import Any, Iterator
 
 import httpx
@@ -36,10 +37,7 @@ def _clear_pool_state() -> None:
         _base_mod._shared_sync_client = None
         _base_mod._shared_sync_refcount = 0
 
-        if _base_mod._shared_async_client is not None and not _base_mod._shared_async_client.is_closed:
-            pass  # async close is best-effort in sync context
-        _base_mod._shared_async_client = None
-        _base_mod._shared_async_refcount = 0
+        _base_mod._async_pools.clear()
 
 
 def _make_client(**kwargs: Any) -> Runloop:
@@ -193,7 +191,7 @@ def test_copy_of_non_shared_stays_non_shared(self):
 
 
 class TestAsyncSharedPool:
-    def test_shared_pool_uses_same_client(self):
+    async def test_shared_pool_uses_same_client(self):
         c1 = _make_async_client(shared_http_pool=True)
         c2 = _make_async_client(shared_http_pool=True)
 
@@ -215,45 +213,80 @@ def test_custom_http_client_bypasses_sharing(self):
         assert c1._client is custom
         assert c1._uses_shared_pool is False
 
-    def test_default_is_shared(self):
+    async def test_default_is_shared(self):
         c1 = _make_async_client()
         assert c1._uses_shared_pool is True
 
 
+def _async_pool_refcount(loop: asyncio.AbstractEventLoop | None) -> int:
+    pool = _base_mod._async_pools.get(loop) if loop is not None else None
+    return pool.refcount if pool is not None else 0
+
+
 class TestAsyncRefcounting:
-    def test_close_one_keeps_pool_alive(self):
+    async def test_close_one_keeps_pool_alive(self):
         c1 = _make_async_client(shared_http_pool=True)
         c2 = _make_async_client(shared_http_pool=True)
+        loop = c1._pool_loop
 
-        # Release c1's ref synchronously (the release_shared_pool is sync)
-        c1._release_shared_pool()
-        assert _base_mod._shared_async_refcount == 1
-        assert _base_mod._shared_async_client is not None
+        c1._release_shared_pool_sync()
+        assert _async_pool_refcount(loop) == 1
 
-        c2._release_shared_pool()
-        assert _base_mod._shared_async_refcount == 0
-        assert _base_mod._shared_async_client is None
+        c2._release_shared_pool_sync()
+        assert _async_pool_refcount(loop) == 0
 
-    def test_double_release_is_safe(self):
+    async def test_double_release_is_safe(self):
         c1 = _make_async_client(shared_http_pool=True)
-        c1._release_shared_pool()
-        c1._release_shared_pool()  # should not raise or go negative
-        assert _base_mod._shared_async_refcount == 0
+        c1._release_shared_pool_sync()
+        c1._release_shared_pool_sync()  # should not raise or go negative
+        assert _async_pool_refcount(c1._pool_loop) == 0
 
 
 class TestAsyncCopy:
-    def test_copy_inherits_shared_pool(self):
+    async def test_copy_inherits_shared_pool(self):
         c1 = _make_async_client(shared_http_pool=True)
         c2 = c1.copy()
+        loop = c1._pool_loop
 
         assert c2._uses_shared_pool is True
         assert c2._client is c1._client
-        assert _base_mod._shared_async_refcount == 2
+        assert _async_pool_refcount(loop) == 2
 
-    def test_copy_with_custom_client_disables_sharing(self):
+    async def test_copy_with_custom_client_disables_sharing(self):
         c1 = _make_async_client(shared_http_pool=True)
         custom = httpx.AsyncClient()
         c2 = c1.copy(http_client=custom)
 
         assert c2._uses_shared_pool is False
         assert c2._client is custom
+
+
+class TestAsyncCrossLoop:
+    def test_separate_loops_get_separate_pools(self):
+        """Clients created in different asyncio.run() calls must not share a pool."""
+
+        async def create_client() -> int:
+            c = _make_async_client(shared_http_pool=True)
+            client_id = id(c._client)
+            await c.close()
+            return client_id
+
+        id1 = asyncio.run(create_client())
+        id2 = asyncio.run(create_client())
+
+        assert id1 != id2, "each loop should get its own httpx.AsyncClient"
+
+    def test_same_loop_shares_pool(self):
+        """Clients created in the same asyncio.run() must share a pool."""
+
+        async def create_two() -> tuple[int, int]:
+            c1 = _make_async_client(shared_http_pool=True)
+            c2 = _make_async_client(shared_http_pool=True)
+            id1 = id(c1._client)
+            id2 = id(c2._client)
+            await c1.close()
+            await c2.close()
+            return id1, id2
+
+        id1, id2 = asyncio.run(create_two())
+        assert id1 == id2