feat(client,transport): expose httpx connection-pool limits on Comfy/AsyncComfy

CHANGELOG.md

@@ -12,6 +12,12 @@ the fuller account of each version, including verification notes.
 
 ### Added
 
+- `Comfy(limits=...)` / `AsyncComfy(limits=...)` — an `httpx.Limits` sizing the connection pool
+  the client's namespaces share. The default is unchanged (httpx's own, 100 connections). Raise
+  `max_connections` when fanning out more than 100 concurrent `models.run` calls: each holds its
+  pooled connection for the whole generation, so past 100 the next call waits on a generation and
+  can end in `httpx.PoolTimeout`. `ComfyLow`/`AsyncComfyLow` take the same keyword; an injected
+  `client=` owns its own pool and ignores it, as it already does `timeout`.
 - `RouterRunResult.credits_used` — what Comfy Router reported a run cost, lifted from the
   `X-Comfy-Credits-Used` response header onto what `models.run_detailed()` returns. It is a
   price rather than a settled ledger entry, absent means "not reported" and never "free", and

README.md

@@ -509,6 +509,23 @@ indefinitely. Each call also sends a fresh `Idempotency-Key`, so an accidental
 exact resend is rejected by the server instead of billing a second generation;
 pass `idempotency_key=` to choose the value yourself.
 
+Holding the connection that long also means holding a *pooled* one: a run
+occupies its slot in the client's httpx connection pool for the whole
+generation. That pool is httpx's default — 100 connections — so a fan-out past
+100 concurrent runs queues the 101st behind a generation finishing, and if
+nothing frees up it waits out the same 10-minute timeout and then raises
+`httpx.PoolTimeout` rather than returning a result. Size the pool up front
+instead of discovering that as a ten-minute stall:
+
+```python
+client = Comfy(limits=httpx.Limits(max_connections=250))
+```
+
+Raising the client's pool is not a licence to exceed the concurrency your
+account is allowed: the server may also bound how many requests a customer has
+in flight at once, and a pool wider than that limit just moves the queue from
+your process to theirs.
+
 ### `models.submit` — queue it, collect it later
 
 `run` holds one connection open until the generation is finished. When the

src/comfy_low/transport.py

@@ -621,12 +621,26 @@ def __init__(
         *,
         client: httpx.Client | None = None,
         timeout: float | None = 30.0,
+        limits: httpx.Limits | None = None,
         client_info: str | None = None,
         router_base_url: str = ROUTER_BASE_URL,
     ) -> None:
+        """``limits`` sizes the connection pool of the client built here.
+
+        ``None`` means httpx's own default (100 connections, 20 kept alive) —
+        it is not forwarded, because ``httpx.Client`` types ``limits`` as a
+        ``Limits`` and does not accept ``None`` for it.
+
+        Passing ``client=`` hands pool ownership to the caller, so ``limits``
+        is ignored on that path exactly as ``timeout`` already is: the injected
+        client was built with its own.
+        """
         self._p = _Prepared(base_url, api_key, client_info, router_base_url)
         self._own_client = client is None
-        self._client = client or httpx.Client(timeout=timeout, follow_redirects=True)
+        kwargs: dict[str, Any] = {"timeout": timeout, "follow_redirects": True}
+        if limits is not None:
+            kwargs["limits"] = limits
+        self._client = client or httpx.Client(**kwargs)
 
     # -- configuration ----------------------------------------------------
     # Read-only views of the settings this transport was built with, so a layer
@@ -1109,12 +1123,22 @@ def __init__(
         *,
         client: httpx.AsyncClient | None = None,
         timeout: float | None = 30.0,
+        limits: httpx.Limits | None = None,
         client_info: str | None = None,
         router_base_url: str = ROUTER_BASE_URL,
     ) -> None:
+        """``limits`` sizes the pool of the client built here — see :class:`ComfyLow`.
+
+        Same two rules: ``None`` means httpx's own default rather than being
+        forwarded, and an injected ``client=`` owns its pool, so ``limits`` is
+        ignored there.
+        """
         self._p = _Prepared(base_url, api_key, client_info, router_base_url)
         self._own_client = client is None
-        self._client = client or httpx.AsyncClient(timeout=timeout, follow_redirects=True)
+        kwargs: dict[str, Any] = {"timeout": timeout, "follow_redirects": True}
+        if limits is not None:
+            kwargs["limits"] = limits
+        self._client = client or httpx.AsyncClient(**kwargs)
 
     # -- configuration (mirrors :class:`ComfyLow`) -------------------------
     @property

src/comfy_sdk/client.py

@@ -43,6 +43,8 @@
 from typing import Any
 from urllib.parse import urlsplit
 
+import httpx
+
 from comfy_low.errors import ApiError
 from comfy_low.transport import ROUTER_BASE_URL, AsyncComfyLow, ComfyLow, origin
 
@@ -230,13 +232,23 @@ class Comfy:
     It does not govern this client's own ``submit``/``run`` (the workflow
     surface), whose 429 handling follows the server's own ``Retry-After``
     instead.
+
+    ``limits`` configures the httpx connection pool every namespace on this
+    client shares; the default is httpx's own (100 connections, 20 kept
+    alive). Raise ``max_connections`` when fanning out more than 100
+    concurrent ``models.run`` calls, because each one holds its pooled
+    connection for the whole generation — past that the next call waits for a
+    generation to finish and eventually fails with ``httpx.PoolTimeout``::
+
+        Comfy(limits=httpx.Limits(max_connections=250))
     """
 
     def __init__(
         self,
         *,
         api_key: str | None = None,
         timeout: float | None = 30.0,
+        limits: httpx.Limits | None = None,
         client_info: str | None = None,
         retry: RetryPolicy = DEFAULT_RETRY,
     ) -> None:
@@ -246,6 +258,7 @@ def __init__(
             base_url,
             key,
             timeout=timeout,
+            limits=limits,
             client_info=client_info,
             router_base_url=_resolve_router_base_url(),
         )
@@ -394,13 +407,22 @@ class AsyncComfy:
     Same credential resolution (explicit ``api_key`` → ``COMFY_API_KEY`` →
     :class:`~comfy_sdk.exceptions.MissingApiKey` on Comfy Cloud) and the same
     key-free ``repr``.
+
+    ``limits`` configures the shared httpx connection pool exactly as on
+    :class:`Comfy`, and matters more here: fanning out past httpx's default of
+    100 connections is easier to do with ``asyncio.gather`` than with threads,
+    and every awaited ``models.run`` holds its connection for the whole
+    generation::
+
+        AsyncComfy(limits=httpx.Limits(max_connections=250))
     """
 
     def __init__(
         self,
         *,
         api_key: str | None = None,
         timeout: float | None = 30.0,
+        limits: httpx.Limits | None = None,
         client_info: str | None = None,
         retry: RetryPolicy = DEFAULT_RETRY,
     ) -> None:
@@ -410,6 +432,7 @@ def __init__(
             base_url,
             key,
             timeout=timeout,
+            limits=limits,
             client_info=client_info,
             router_base_url=_resolve_router_base_url(),
         )

tests/test_models_namespace.py

@@ -12,6 +12,7 @@
 import httpx
 
 import comfy_sdk
+from comfy_low.transport import ComfyLow
 from comfy_sdk import AsyncComfy, Comfy
 from comfy_sdk.models import AsyncModels, Models
 
@@ -70,6 +71,53 @@ async def test_a_timeout_change_on_the_async_client_is_visible_through_models(se
         assert client.models.timeout.read == 1.25
 
 
+# --- the pool the namespace shares --------------------------------------
+#
+# `models.run` holds its pooled connection for the whole generation, so the
+# pool's ceiling is what caps a fan-out of concurrent runs. These reach into
+# httpx/httpcore internals (`_transport._pool._max_connections`) because httpx
+# exposes no public getter for the limits a client was built with; the
+# alternative is asserting nothing about whether `limits=` arrived at all.
+
+
+def test_limits_sizes_the_pool_the_client_builds(server) -> None:
+    with Comfy(limits=httpx.Limits(max_connections=250)) as client:
+        assert client._low._client._transport._pool._max_connections == 250
+
+
+async def test_limits_sizes_the_pool_the_async_client_builds(server) -> None:
+    async with AsyncComfy(limits=httpx.Limits(max_connections=250)) as client:
+        assert client._low._client._transport._pool._max_connections == 250
+
+
+def test_no_limits_leaves_httpxs_own_default_pool(server) -> None:
+    # `limits=None` is dropped rather than forwarded — httpx types the
+    # parameter as a `Limits` and does not accept `None` — so the default path
+    # has to land on httpx's own DEFAULT_LIMITS, 100 connections.
+    with Comfy() as client:
+        assert client._low._client._transport._pool._max_connections == 100
+
+
+async def test_no_limits_leaves_the_async_client_httpxs_default_pool(server) -> None:
+    async with AsyncComfy() as client:
+        assert client._low._client._transport._pool._max_connections == 100
+
+
+def test_an_injected_client_keeps_its_own_pool(server) -> None:
+    # Injecting `client=` hands pool ownership to the caller, so `limits` is
+    # ignored there exactly as `timeout` already is: the pool stays the
+    # injected client's default 100, not the 5 asked for here.
+    with httpx.Client() as injected:
+        low = ComfyLow(
+            server.base_url,
+            "ck_test",
+            client=injected,
+            limits=httpx.Limits(max_connections=5),
+        )
+        assert low._client is injected
+        assert low._client._transport._pool._max_connections == 100
+
+
 def test_models_sends_the_host_clients_credentials(server) -> None:
     server.state.require_auth = True
     with Comfy(api_key="k-first") as client:

tests/test_router_spec_contract.py

@@ -294,11 +294,15 @@ def test_the_bound_path_has_exactly_the_two_segments_the_binding_fills() -> None
     "dropped_params": "X-Comfy-Router-Dropped-Params",
     "replayed": "Idempotent-Replayed",
     "request_id": "X-Comfy-Request-Id",
+    "credits_used": "X-Comfy-Credits-Used",
 }
 
-#: Lifted by the SDK but NOT declared on the contract's 200 -- see the tripwire
-#: test at the bottom of this file.
-_UNDECLARED_HEADER_LIFTS = {"credits_used": "X-Comfy-Credits-Used"}
+#: Header value to use for the "present" case in
+#: test_the_lift_actually_reads_the_declared_name. "x" round-trips fine for
+#: the pass-through lifts, but credits_used parses its header as a decimal,
+#: so a non-numeric sample would normalise to None same as absent and the
+#: test would pass for the wrong reason.
+_SAMPLE_HEADER_VALUES: dict[str, str] = {"credits_used": "1.25"}
 
 
 def _declared_run_response_headers() -> set[str]:
@@ -333,34 +337,9 @@ def test_the_lift_actually_reads_the_declared_name(field: str, header: str) -> N
     to fail.
     """
     absent = getattr(_run_result({}, {}), field)
-    present = getattr(_run_result({}, {header: "x"}), field)
+    sample = _SAMPLE_HEADER_VALUES.get(field, "x")
+    present = getattr(_run_result({}, {header: sample}), field)
     assert present != absent, (
         f"_run_result ignored {header!r}: RouterRunResult.{field} read {absent!r} both with "
         f"the header and without it, so the lift is reading some other name."
     )
-
-
-@pytest.mark.parametrize(("field", "header"), sorted(_UNDECLARED_HEADER_LIFTS.items()))
-def test_an_undeclared_lift_stays_undeclared_until_someone_reconciles_it(
-    field: str, header: str
-) -> None:
-    """Tripwire, and deliberately asserting the *absence*.
-
-    ``credits_used`` is lifted from a header the vendored contract does not
-    declare anywhere -- the 200's only cost headers are the
-    ``X-Committed-Spend-*`` trio, which is a different quantity (USD cents of
-    in-flight commitment, not the price of this run). Nothing in the suite can
-    catch a wrong name here, because every test configures its stub to emit the
-    exact literal the lift reads.
-
-    That gap is tracked, not accepted. This test fails the moment a spec sync
-    declares the header, which is the signal to move the entry up into
-    ``_CONTRACT_HEADER_LIFTS`` and get it pinned like the rest. It also fails
-    if the header is declared under a *different* name for the same quantity,
-    because the reconciliation is the same either way.
-    """
-    declared = _declared_run_response_headers()
-    assert header not in declared, (
-        f"the vendored spec now declares {header!r}: move {field!r} from "
-        f"_UNDECLARED_HEADER_LIFTS into _CONTRACT_HEADER_LIFTS so it is pinned."
-    )