fix: open extra H2 conns under stream pressure; isolate file transfers on HTTP/1.1

httpcore blocks on a per-connection stream semaphore once slots fill instead of
opening another connection. Patch is_available + non-blocking acquire so the pool
spreads load. Route upload_file/download_file through a dedicated HTTP/1.1 pool so
bulk bodies cannot starve latency-sensitive RPCs on the shared H2 session.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: Reflex

src/runloop_api_client/__init__.py

@@ -28,8 +28,13 @@
     APIResponseValidationError,
 )
 from ._base_client import DefaultHttpxClient, DefaultAioHttpClient, DefaultAsyncHttpxClient
+from ._http2_pool_fix import install as _install_http2_pool_fix
 from ._utils._logs import setup_logging as _setup_logging
 
+# Open additional HTTP/2 connections when a connection's stream slots are full
+# instead of blocking on httpcore's per-connection stream semaphore.
+_install_http2_pool_fix()
+
 __all__ = [
     "types",
     "__version__",

src/runloop_api_client/_base_client.py

@@ -174,6 +174,21 @@ async def aclose(self) -> None:
     weakref.WeakKeyDictionary()
 )
 
+# Separate HTTP/1.1 pool for bulk file upload/download. Keeping these off the
+# main HTTP/2 connection avoids H2 session-window / stream-slot contention with
+# latency-sensitive RPCs (create, wait_for_status, execute, …).
+_shared_sync_transfer_transport: _SharedTransport | None = None
+_shared_async_transfer_transports: weakref.WeakKeyDictionary[
+    asyncio.AbstractEventLoop, _SharedAsyncTransport
+] = weakref.WeakKeyDictionary()
+
+# Paths that carry large request/response bodies and should use the transfer pool.
+_FILE_TRANSFER_PATH_SUFFIXES = ("/upload_file", "/download_file")
+
+
+def _is_file_transfer_path(path: str) -> bool:
+    return path.endswith(_FILE_TRANSFER_PATH_SUFFIXES)
+
 # TODO: make base page type vars covariant
 SyncPageT = TypeVar("SyncPageT", bound="BaseSyncPage[Any]")
 AsyncPageT = TypeVar("AsyncPageT", bound="BaseAsyncPage[Any]")
@@ -929,8 +944,10 @@ def __del__(self) -> None:
 
 class SyncAPIClient(BaseClient[httpx.Client, Stream[Any]]):
     _client: httpx.Client
+    _transfer_client: httpx.Client | None
     _default_stream_cls: type[Stream[Any]] | None = None
     _uses_shared_pool: bool
+    _isolate_file_transfers: bool
     _closed: bool
 
     def __init__(
@@ -976,6 +993,9 @@ def __init__(
         )
 
         self._closed = False
+        self._transfer_client = None
+        # Custom http_client owns the full transport stack; don't invent a sibling pool.
+        self._isolate_file_transfers = http_client is None
 
         if http_client is not None:
             self._client = http_client
@@ -1000,6 +1020,38 @@ def __init__(
             )
             self._uses_shared_pool = False
 
+    def _ensure_transfer_client(self) -> httpx.Client:
+        """Lazy HTTP/1.1 client for upload_file / download_file."""
+        if self._transfer_client is not None:
+            return self._transfer_client
+
+        timeout = cast(Timeout, self.timeout)
+        if self._uses_shared_pool:
+            global _shared_sync_transfer_transport
+            with _pool_lock:
+                if _shared_sync_transfer_transport is None or not _shared_sync_transfer_transport.acquire():
+                    _shared_sync_transfer_transport = _SharedTransport(
+                        httpx.HTTPTransport(limits=DEFAULT_CONNECTION_LIMITS, http2=False),
+                    )
+            self._transfer_client = SyncHttpxClientWrapper(
+                base_url=self._base_url,
+                timeout=timeout,
+                transport=_shared_sync_transfer_transport,
+                http2=False,
+            )
+        else:
+            self._transfer_client = SyncHttpxClientWrapper(
+                base_url=self._base_url,
+                timeout=timeout,
+                http2=False,
+            )
+        return self._transfer_client
+
+    def _send_client_for_request(self, request: httpx.Request) -> httpx.Client:
+        if self._isolate_file_transfers and _is_file_transfer_path(request.url.path):
+            return self._ensure_transfer_client()
+        return self._client
+
     def is_closed(self) -> bool:
         return self._closed or self._client.is_closed
 
@@ -1014,6 +1066,10 @@ def close(self) -> None:
             return
         self._closed = True
         self._client.close()
+        transfer = self._transfer_client
+        self._transfer_client = None
+        if transfer is not None:
+            transfer.close()
 
     def __enter__(self: _T) -> _T:
         return self
@@ -1114,7 +1170,7 @@ def request(
 
             response = None
             try:
-                response = self._client.send(
+                response = self._send_client_for_request(request).send(
                     request,
                     stream=stream or self._should_stream_response_body(request=request),
                     **kwargs,
@@ -1561,8 +1617,10 @@ def __del__(self) -> None:
 
 class AsyncAPIClient(BaseClient[httpx.AsyncClient, AsyncStream[Any]]):
     _client: httpx.AsyncClient
+    _transfer_client: httpx.AsyncClient | None
     _default_stream_cls: type[AsyncStream[Any]] | None = None
     _uses_shared_pool: bool
+    _isolate_file_transfers: bool
     _closed: bool
 
     def __init__(
@@ -1608,6 +1666,9 @@ def __init__(
         )
 
         self._closed = False
+        self._transfer_client = None
+        # Custom http_client owns the full transport stack; don't invent a sibling pool.
+        self._isolate_file_transfers = http_client is None
 
         if http_client is not None:
             self._client = http_client
@@ -1646,6 +1707,47 @@ def __init__(
             )
             self._uses_shared_pool = False
 
+    def _ensure_transfer_client(self) -> httpx.AsyncClient:
+        """Lazy HTTP/1.1 client for upload_file / download_file."""
+        if self._transfer_client is not None:
+            return self._transfer_client
+
+        timeout = cast(Timeout, self.timeout)
+        if self._uses_shared_pool:
+            try:
+                loop: asyncio.AbstractEventLoop | None = asyncio.get_running_loop()
+            except RuntimeError:
+                loop = None
+            if loop is not None:
+                with _pool_lock:
+                    existing = _shared_async_transfer_transports.get(loop)
+                    if existing is not None and existing.acquire():
+                        transport: _SharedAsyncTransport = existing
+                    else:
+                        transport = _SharedAsyncTransport(
+                            httpx.AsyncHTTPTransport(limits=DEFAULT_CONNECTION_LIMITS, http2=False),
+                        )
+                        _shared_async_transfer_transports[loop] = transport
+                self._transfer_client = AsyncHttpxClientWrapper(
+                    base_url=self._base_url,
+                    timeout=timeout,
+                    transport=transport,
+                    http2=False,
+                )
+                return self._transfer_client
+
+        self._transfer_client = AsyncHttpxClientWrapper(
+            base_url=self._base_url,
+            timeout=timeout,
+            http2=False,
+        )
+        return self._transfer_client
+
+    def _send_client_for_request(self, request: httpx.Request) -> httpx.AsyncClient:
+        if self._isolate_file_transfers and _is_file_transfer_path(request.url.path):
+            return self._ensure_transfer_client()
+        return self._client
+
     def is_closed(self) -> bool:
         return self._closed or self._client.is_closed
 
@@ -1660,6 +1762,10 @@ async def close(self) -> None:
             return
         self._closed = True
         await self._client.aclose()
+        transfer = self._transfer_client
+        self._transfer_client = None
+        if transfer is not None:
+            await transfer.aclose()
 
     async def __aenter__(self: _T) -> _T:
         return self
@@ -1765,7 +1871,7 @@ async def request(
 
             response = None
             try:
-                response = await self._client.send(
+                response = await self._send_client_for_request(request).send(
                     request,
                     stream=stream or self._should_stream_response_body(request=request),
                     **kwargs,

src/runloop_api_client/_http2_pool_fix.py

@@ -0,0 +1,185 @@
+"""Make httpcore open additional HTTP/2 connections when stream slots run low.
+
+httpcore multiplexes HTTP/2 on a single connection and gates new streams with a
+semaphore (default MAX_CONCURRENT_STREAMS=100).  Its connection pool's
+``is_available()`` ignores that limit, so once stream slots are exhausted,
+further requests *block* on the semaphore instead of the pool opening another
+connection — even when ``max_connections`` still has headroom.
+
+Additionally, the pool may assign a *burst* of requests to a connection that
+only has a few free slots left (``is_available()`` is checked once per request
+but slots are not reserved).  Winners then share an overloaded connection while
+the rest raise ``ConnectionNotAvailable`` and retry.
+
+This module patches sync + async HTTP/2 connections so that:
+
+1. ``is_available()`` is False when free stream slots are below a headroom
+   threshold, so the pool prefers opening/reusing another connection.
+2. Stream-slot acquire is non-blocking; if a race still over-assigns, we raise
+   ``ConnectionNotAvailable`` and the pool retries on another connection.
+
+Idempotent: safe to call ``install()`` more than once.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Callable
+
+logger = logging.getLogger("runloop_api_client._http2_pool_fix")
+
+# If a connection has fewer free stream slots than this, treat it as unavailable
+# so the pool opens another connection instead of stampeding the remainder.
+# Must be >1: the pool assigns many queued requests to one "available" conn
+# without reserving slots, so small remaining capacity still over-assigns.
+_MIN_FREE_STREAM_SLOTS = 16
+
+_installed = False
+
+
+def _free_stream_slots(connection: Any) -> int | None:
+    """Return free H2 stream slots, or None if unknown."""
+    max_streams = getattr(connection, "_max_streams", None)
+    events = getattr(connection, "_events", None)
+    if not isinstance(max_streams, int) or max_streams <= 0 or events is None:
+        return None
+    return max_streams - len(events)
+
+
+def _stream_slots_saturated(connection: Any) -> bool:
+    """True when the connection should not accept more streams."""
+    free = _free_stream_slots(connection)
+    if free is None:
+        return False
+    return free < _MIN_FREE_STREAM_SLOTS
+
+
+def _patch_is_available(cls: type) -> None:
+    if getattr(cls.is_available, "_runloop_stream_overflow_patched", False):
+        return
+
+    original: Callable[[Any], bool] = cls.is_available
+
+    def is_available(self: Any) -> bool:
+        if not original(self):
+            return False
+        return not _stream_slots_saturated(self)
+
+    is_available._runloop_stream_overflow_patched = True  # type: ignore[attr-defined]
+    cls.is_available = is_available  # type: ignore[method-assign]
+
+
+def _make_async_nonblocking_acquire(sem: Any, connection_not_available: type) -> Callable[[], Any]:
+    original_acquire = sem.acquire
+
+    async def acquire() -> None:
+        if not getattr(sem, "_backend", ""):
+            sem.setup()
+
+        if sem._backend == "asyncio":
+            inner = sem._anyio_semaphore
+            try:
+                inner.acquire_nowait()
+                return
+            except Exception:
+                raise connection_not_available(
+                    "HTTP/2 connection has no free stream slots"
+                ) from None
+
+        if sem._backend == "trio":
+            inner = sem._trio_semaphore
+            acquire_nowait = getattr(inner, "acquire_nowait", None)
+            if acquire_nowait is not None:
+                try:
+                    acquire_nowait()
+                    return
+                except Exception:
+                    raise connection_not_available(
+                        "HTTP/2 connection has no free stream slots"
+                    ) from None
+
+        await original_acquire()
+
+    return acquire
+
+
+def _make_sync_nonblocking_acquire(sem: Any, connection_not_available: type) -> Callable[[], None]:
+    inner = getattr(sem, "_semaphore", sem)
+
+    def acquire() -> None:
+        ok = inner.acquire(False)
+        if not ok:
+            raise connection_not_available("HTTP/2 connection has no free stream slots")
+
+    return acquire
+
+
+def _patch_async_connection(cls: type) -> None:
+    from httpcore import ConnectionNotAvailable
+
+    if getattr(cls.handle_async_request, "_runloop_stream_overflow_patched", False):
+        return
+
+    original = cls.handle_async_request
+
+    async def handle_async_request(self: Any, request: Any) -> Any:
+        sem = getattr(self, "_max_streams_semaphore", None)
+        if sem is None:
+            return await original(self, request)
+
+        real_acquire = sem.acquire
+        sem.acquire = _make_async_nonblocking_acquire(sem, ConnectionNotAvailable)
+        try:
+            return await original(self, request)
+        finally:
+            sem.acquire = real_acquire
+
+    handle_async_request._runloop_stream_overflow_patched = True  # type: ignore[attr-defined]
+    cls.handle_async_request = handle_async_request  # type: ignore[method-assign]
+
+
+def _patch_sync_connection(cls: type) -> None:
+    from httpcore import ConnectionNotAvailable
+
+    if getattr(cls.handle_request, "_runloop_stream_overflow_patched", False):
+        return
+
+    original = cls.handle_request
+
+    def handle_request(self: Any, request: Any) -> Any:
+        sem = getattr(self, "_max_streams_semaphore", None)
+        if sem is None:
+            return original(self, request)
+
+        real_acquire = sem.acquire
+        sem.acquire = _make_sync_nonblocking_acquire(sem, ConnectionNotAvailable)
+        try:
+            return original(self, request)
+        finally:
+            sem.acquire = real_acquire
+
+    handle_request._runloop_stream_overflow_patched = True  # type: ignore[attr-defined]
+    cls.handle_request = handle_request  # type: ignore[method-assign]
+
+
+def install() -> bool:
+    """Patch httpcore HTTP/2 connections. Returns True if a patch was applied."""
+    global _installed
+    if _installed:
+        return False
+
+    try:
+        from httpcore._async.http2 import AsyncHTTP2Connection
+        from httpcore._sync.http2 import HTTP2Connection
+    except ImportError as exc:  # pragma: no cover
+        logger.warning("http2 stream-overflow patch skipped: %s", exc)
+        return False
+
+    _patch_is_available(AsyncHTTP2Connection)
+    _patch_is_available(HTTP2Connection)
+    _patch_async_connection(AsyncHTTP2Connection)
+    _patch_sync_connection(HTTP2Connection)
+
+    _installed = True
+    logger.debug("installed httpcore HTTP/2 stream-overflow connection patch")
+    return True