fix(python-sdk): key async transport caches by loop object, not id(loop) (#1434)

## Description

The per-event-loop caches for `AsyncHTTPTransport`s and
`httpx.AsyncClient`s (Sandbox API, envd, and volume clients) were keyed
by `id(asyncio.get_running_loop())`, but CPython reuses object ids of
dead loops almost immediately — so sequential loops could inherit a
transport bound to a previous, closed loop and fail. The caches are now
`weakref.WeakKeyDictionary`s keyed by the loop object itself, which
makes stale id collisions impossible and releases entries when their
loop is garbage collected (fixing a leak where dead-loop entries
accumulated forever). Added regression tests covering the
sequential-loop scenario for all three caches.

This pattern no longer breaks:

```python
# e.g. a worker or test harness running repeated event loops
for job in jobs:
    asyncio.run(process_with_sandbox(job))  # each run previously risked
                                            # inheriting a closed loop's transport
```

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: mishushakov

.changeset/eager-loops-retire.md

@@ -0,0 +1,5 @@
+---
+"@e2b/python-sdk": patch
+---
+
+Key per-event-loop async HTTP transport and client caches by the loop object (held weakly) instead of `id(loop)`. CPython can reuse the id of a closed loop almost immediately, so sequential event loops (for example repeated `asyncio.run(...)` calls) could inherit a transport or client bound to a previous, closed loop. Cache entries are now also released automatically when their loop is garbage collected.

packages/python-sdk/e2b/api/__init__.py

@@ -4,9 +4,10 @@
 import os
 import re
 import threading
+import weakref
 from dataclasses import dataclass
 from types import TracebackType
-from typing import Callable, Dict, Optional, Protocol, Union
+from typing import Callable, Optional, Protocol, Union
 
 import httpx
 from httpx import AsyncBaseTransport, BaseTransport, Limits, Timeout
@@ -120,7 +121,12 @@ def __init__(
         self._transport_factory = transport_factory
         self._async_transport_factory = async_transport_factory
         self._thread_local = threading.local()
-        self._async_clients: Dict[int, httpx.AsyncClient] = {}
+        # Keyed weakly by the event loop object itself, not id(loop) —
+        # CPython reuses object ids, so a new loop could otherwise inherit
+        # a client bound to a previous, closed loop.
+        self._async_clients: weakref.WeakKeyDictionary[
+            asyncio.AbstractEventLoop, httpx.AsyncClient
+        ] = weakref.WeakKeyDictionary()
         self._proxy = config.proxy
 
         if require_api_key and require_access_token:
@@ -231,8 +237,8 @@ def get_async_httpx_client(self) -> httpx.AsyncClient:
         if self._async_client is not None or self._async_transport_factory is None:
             return super().get_async_httpx_client()
 
-        loop_id = id(asyncio.get_running_loop())
-        client = self._async_clients.get(loop_id)
+        loop = asyncio.get_running_loop()
+        client = self._async_clients.get(loop)
         if client is None:
             client = httpx.AsyncClient(
                 base_url=self._base_url,
@@ -244,7 +250,7 @@ def get_async_httpx_client(self) -> httpx.AsyncClient:
                 event_hooks=self._httpx_args.get("event_hooks"),
                 transport=self._async_transport_factory(),
             )
-            self._async_clients[loop_id] = client
+            self._async_clients[loop] = client
         return client
 
     def _log_request(self, request):

packages/python-sdk/e2b/api/client_async/__init__.py

@@ -1,5 +1,6 @@
 import asyncio
 import logging
+import weakref
 from typing import Dict, Optional, Tuple
 
 import httpx
@@ -11,7 +12,7 @@
 
 logger = logging.getLogger(__name__)
 
-TransportKey = Tuple[int, bool, Optional[ProxyTypes]]
+TransportKey = Tuple[bool, Optional[ProxyTypes]]
 
 
 def get_api_client(config: ConnectionConfig, **kwargs) -> AsyncApiClient:
@@ -23,7 +24,13 @@ def get_api_client(config: ConnectionConfig, **kwargs) -> AsyncApiClient:
 
 
 class AsyncTransportWithLogger(httpx.AsyncHTTPTransport):
-    _instances: Dict[TransportKey, "AsyncTransportWithLogger"] = {}
+    # Keyed weakly by the event loop object itself, not id(loop) — CPython
+    # reuses object ids, so a new loop could otherwise inherit a transport
+    # bound to a previous, closed loop.
+    _instances: weakref.WeakKeyDictionary[
+        asyncio.AbstractEventLoop,
+        Dict[TransportKey, "AsyncTransportWithLogger"],
+    ] = weakref.WeakKeyDictionary()
 
     async def handle_async_request(self, request):
         url = f"{request.url.scheme}://{request.url.host}{request.url.path}"
@@ -40,43 +47,41 @@ def pool(self):
         return self._pool
 
 
-def get_transport(
-    config: ConnectionConfig, http2: bool = True
-) -> AsyncTransportWithLogger:
-    key: TransportKey = (id(asyncio.get_running_loop()), http2, config.proxy)
+def _get_cached_transport(cls, config: ConnectionConfig, http2: bool):
+    loop = asyncio.get_running_loop()
+    loop_instances = cls._instances.get(loop)
+    if loop_instances is None:
+        loop_instances = {}
+        cls._instances[loop] = loop_instances
+
+    key: TransportKey = (http2, config.proxy)
+    transport = loop_instances.get(key)
+    if transport is None:
+        transport = cls(
+            limits=limits,
+            proxy=config.proxy,
+            http2=http2,
+            retries=connection_retries,
+        )
+        loop_instances[key] = transport
 
-    if key in AsyncTransportWithLogger._instances:
-        return AsyncTransportWithLogger._instances[key]
+    return transport
 
-    transport = AsyncTransportWithLogger(
-        limits=limits,
-        proxy=config.proxy,
-        http2=http2,
-        retries=connection_retries,
-    )
 
-    AsyncTransportWithLogger._instances[key] = transport
-    return transport
+def get_transport(
+    config: ConnectionConfig, http2: bool = True
+) -> AsyncTransportWithLogger:
+    return _get_cached_transport(AsyncTransportWithLogger, config, http2)
 
 
 class AsyncEnvdTransportWithLogger(AsyncTransportWithLogger):
-    _instances: Dict[TransportKey, "AsyncEnvdTransportWithLogger"] = {}
+    _instances: weakref.WeakKeyDictionary[
+        asyncio.AbstractEventLoop,
+        Dict[TransportKey, "AsyncEnvdTransportWithLogger"],
+    ] = weakref.WeakKeyDictionary()
 
 
 def get_envd_transport(
     config: ConnectionConfig, http2: bool = True
 ) -> AsyncEnvdTransportWithLogger:
-    key: TransportKey = (id(asyncio.get_running_loop()), http2, config.proxy)
-
-    if key in AsyncEnvdTransportWithLogger._instances:
-        return AsyncEnvdTransportWithLogger._instances[key]
-
-    transport = AsyncEnvdTransportWithLogger(
-        limits=limits,
-        proxy=config.proxy,
-        http2=http2,
-        retries=connection_retries,
-    )
-
-    AsyncEnvdTransportWithLogger._instances[key] = transport
-    return transport
+    return _get_cached_transport(AsyncEnvdTransportWithLogger, config, http2)

packages/python-sdk/e2b/volume/client_async/__init__.py

@@ -1,7 +1,8 @@
 import asyncio
 import logging
 import os
-from typing import Dict, Optional, Tuple
+import weakref
+from typing import Dict, Optional
 
 import httpx
 from httpx import Limits
@@ -20,7 +21,7 @@
     keepalive_expiry=int(os.getenv("E2B_KEEPALIVE_EXPIRY", "300")),
 )
 
-TransportKey = Tuple[int, Optional[ProxyTypes]]
+TransportKey = Optional[ProxyTypes]
 
 
 def get_api_client(config: VolumeConnectionConfig, **kwargs) -> AsyncVolumeApiClient:
@@ -53,7 +54,13 @@ def get_api_client(config: VolumeConnectionConfig, **kwargs) -> AsyncVolumeApiCl
 
 
 class AsyncTransportWithLogger(httpx.AsyncHTTPTransport):
-    _instances: Dict[TransportKey, "AsyncTransportWithLogger"] = {}
+    # Keyed weakly by the event loop object itself, not id(loop) — CPython
+    # reuses object ids, so a new loop could otherwise inherit a transport
+    # bound to a previous, closed loop.
+    _instances: weakref.WeakKeyDictionary[
+        asyncio.AbstractEventLoop,
+        Dict[TransportKey, "AsyncTransportWithLogger"],
+    ] = weakref.WeakKeyDictionary()
 
     async def handle_async_request(self, request):
         url = f"{request.url.scheme}://{request.url.host}{request.url.path}"
@@ -68,14 +75,19 @@ def pool(self):
 
 
 def get_transport(config: VolumeConnectionConfig) -> AsyncTransportWithLogger:
-    key: TransportKey = (id(asyncio.get_running_loop()), config.proxy)
-
-    if key in AsyncTransportWithLogger._instances:
-        return AsyncTransportWithLogger._instances[key]
+    loop = asyncio.get_running_loop()
+    loop_instances = AsyncTransportWithLogger._instances.get(loop)
+    if loop_instances is None:
+        loop_instances = {}
+        AsyncTransportWithLogger._instances[loop] = loop_instances
+
+    key: TransportKey = config.proxy
+    transport = loop_instances.get(key)
+    if transport is None:
+        transport = AsyncTransportWithLogger(
+            limits=limits,
+            proxy=config.proxy,
+        )
+        loop_instances[key] = transport
 
-    transport = AsyncTransportWithLogger(
-        limits=limits,
-        proxy=config.proxy,
-    )
-    AsyncTransportWithLogger._instances[key] = transport
     return transport

packages/python-sdk/tests/test_api_client_transport.py

@@ -1,4 +1,5 @@
 import asyncio
+import gc
 from concurrent.futures import ThreadPoolExecutor
 
 import httpx
@@ -222,8 +223,8 @@ async def test_async_api_client_proxy_uses_explicit_transport(test_api_key):
 
     api_client = get_async_api_client(config)
     httpx_client = api_client.get_async_httpx_client()
-    transport = AsyncTransportWithLogger._instances[
-        (id(asyncio.get_running_loop()), True, "http://127.0.0.1:9999")
+    transport = AsyncTransportWithLogger._instances[asyncio.get_running_loop()][
+        (True, "http://127.0.0.1:9999")
     ]
 
     try:
@@ -335,6 +336,72 @@ async def get_client_ids():
         AsyncTransportWithLogger._instances.clear()
 
 
+def test_async_transport_not_reused_across_sequential_loops(test_api_key):
+    AsyncTransportWithLogger._instances.clear()
+    config = ConnectionConfig(api_key=test_api_key)
+
+    async def get_transport():
+        return get_async_transport(config)
+
+    try:
+        loop_a = asyncio.new_event_loop()
+        try:
+            transport_a = loop_a.run_until_complete(get_transport())
+        finally:
+            loop_a.close()
+        del loop_a
+        gc.collect()
+
+        # The cache entry dies with the loop, so a later loop can never
+        # inherit a transport bound to a closed loop, even when CPython
+        # reuses the dead loop's object id.
+        assert len(AsyncTransportWithLogger._instances) == 0
+
+        loop_b = asyncio.new_event_loop()
+        try:
+            transport_b = loop_b.run_until_complete(get_transport())
+        finally:
+            loop_b.close()
+
+        assert transport_b is not transport_a
+    finally:
+        AsyncTransportWithLogger._instances.clear()
+
+
+def test_async_api_client_not_reused_across_sequential_loops(test_api_key):
+    AsyncTransportWithLogger._instances.clear()
+    config = ConnectionConfig(api_key=test_api_key)
+    api_client = get_async_api_client(config)
+
+    async def get_client():
+        client = api_client.get_async_httpx_client()
+        assert api_client.get_async_httpx_client() is client
+        return client
+
+    try:
+        loop_a = asyncio.new_event_loop()
+        try:
+            client_a = loop_a.run_until_complete(get_client())
+            loop_a.run_until_complete(client_a.aclose())
+        finally:
+            loop_a.close()
+        del loop_a
+        gc.collect()
+
+        assert len(api_client._async_clients) == 0
+
+        loop_b = asyncio.new_event_loop()
+        try:
+            client_b = loop_b.run_until_complete(get_client())
+            loop_b.run_until_complete(client_b.aclose())
+        finally:
+            loop_b.close()
+
+        assert client_b is not client_a
+    finally:
+        AsyncTransportWithLogger._instances.clear()
+
+
 @pytest.mark.asyncio
 async def test_async_envd_transport_uses_separate_cache(test_api_key):
     AsyncTransportWithLogger._instances.clear()

packages/python-sdk/tests/test_volume_client.py

@@ -1,11 +1,13 @@
 import asyncio
+import gc
 import threading
 
 import httpx
 import pytest
 
 from e2b.exceptions import AuthenticationException
 from e2b.volume.client_async import (
+    AsyncTransportWithLogger as AsyncVolumeTransport,
     get_api_client as get_async_api_client,
     get_transport as get_async_transport,
 )
@@ -96,7 +98,6 @@ async def get_transports():
     async def get_proxied_transport():
         return get_async_transport(proxied)
 
-    # Keep both loops alive at the same time so their ids cannot collide
     loop_a = asyncio.new_event_loop()
     loop_b = asyncio.new_event_loop()
     try:
@@ -113,3 +114,32 @@ async def get_proxied_transport():
     finally:
         loop_a.close()
         loop_b.close()
+
+
+def test_async_transport_not_reused_across_sequential_loops():
+    AsyncVolumeTransport._instances.clear()
+    config = VolumeConnectionConfig(token="vol-token")
+
+    async def get_transport():
+        return get_async_transport(config)
+
+    loop_a = asyncio.new_event_loop()
+    try:
+        transport_a = loop_a.run_until_complete(get_transport())
+    finally:
+        loop_a.close()
+    del loop_a
+    gc.collect()
+
+    # The cache entry dies with the loop, so a later loop can never inherit
+    # a transport bound to a closed loop, even when CPython reuses the dead
+    # loop's object id.
+    assert len(AsyncVolumeTransport._instances) == 0
+
+    loop_b = asyncio.new_event_loop()
+    try:
+        transport_b = loop_b.run_until_complete(get_transport())
+    finally:
+        loop_b.close()
+
+    assert transport_b is not transport_a