fix: bound the response body by the total on the httpx family

clientwright/adapters/_httpx_shared.py

@@ -13,6 +13,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import ssl
 from collections.abc import Callable, Mapping, MutableMapping
 from typing import Any
@@ -32,6 +33,7 @@
 from ..core.model import IDEMPOTENT_METHODS, ConnMetrics, FailureKind, Outcome, RequestInfo, ResolvedTimeouts, origin_of
 from ..core.native import accepted_overrides, validate_native
 from ..core.plan import CallPlan, ClientHandle, ClientRuntime, compile_plan, register_handle
+from ..core.policy.budget import Deadline
 from ..core.policy.timeout import base_timeouts
 from ..core.telemetry.emitter import ClientTelemetry
 
@@ -84,6 +86,7 @@ def capabilities_for(adapter: str) -> AdapterCapabilities:
             Capability.TIMEOUT_WRITE: Support.NATIVE,
             Capability.TIMEOUT_POOL: Support.NATIVE,
             Capability.DEADLINE_HARD: Support.EMULATED,
+            Capability.DEADLINE_COVERS_BODY: Support.EMULATED,
             Capability.POOL_LIMIT_TOTAL: Support.NATIVE,
             Capability.POOL_LIMIT_PER_HOST: Support.EMULATED,
             Capability.KEEPALIVE: Support.NATIVE,
@@ -123,6 +126,10 @@ def capabilities_for(adapter: str) -> AdapterCapabilities:
             "pool_limit_per_host": "Emulated as a per-origin in-flight semaphore; limits requests, not connections.",
             "dns_error": f"{adapter} wraps DNS failures into ConnectError; they surface as connect_error.",
             "proxy_from_env": "Environment proxies are parsed into mounts; NO_PROXY entries match hosts literally.",
+            "deadline_covers_body": (
+                "The response body is inside the total: the async client bounds every chunk by the remaining budget, "
+                "the sync client refuses the next chunk once the budget is gone (soft)."
+            ),
         },
     )
 
@@ -140,8 +147,6 @@ def has_ssl_cause(exc: BaseException) -> bool:
 
 def classify_family_error(sdk: Any, exc: BaseException) -> FailureKind:
     """Exception -> FailureKind for any SDK exposing the httpx error names."""
-    import asyncio  # noqa: PLC0415 - stdlib, deferred to keep module import light
-
     if isinstance(exc, asyncio.CancelledError):
         return FailureKind.CANCELLED
     if isinstance(exc, sdk.ConnectTimeout):
@@ -344,12 +349,21 @@ def location(self) -> str | None:
 
 
 class TimedStreamCore:
-    """Times body consumption and reports read failures exactly once."""
+    """Times body consumption, keeps it inside the deadline and reports the end exactly once."""
 
-    def __init__(self, inner: Any, clock: Callable[[], float], on_done: Callable[[Outcome, float], None]) -> None:
+    def __init__(
+        self,
+        inner: Any,
+        clock: Callable[[], float],
+        on_done: Callable[[Outcome, float], None],
+        deadline: Deadline,
+        translate: Callable[[CallError], BaseException],
+    ) -> None:
         self._inner = inner
         self._clock = clock
         self._on_done = on_done
+        self._deadline = deadline
+        self._translate = translate
         self._started = clock()
         self._finished = False
 
@@ -358,14 +372,37 @@ def _finish(self, outcome: Outcome) -> None:
             self._finished = True
             self._on_done(outcome, self._clock() - self._started)
 
+    def _failed(self, exc: BaseException) -> None:
+        kind = FailureKind.TOTAL_TIMEOUT if isinstance(exc, DeadlineExceededError) else FailureKind.BODY_ERROR
+        self._finish(Outcome(kind=kind, exception=exc))
+
+    def _expired(self) -> BaseException:
+        return self._translate(DeadlineExceededError(self._deadline.total or 0.0))
+
+    def _check_budget(self) -> None:
+        if self._deadline.expired:
+            raise self._expired()
+
 
 class AsyncTimedStreamMixin(TimedStreamCore):
     async def __aiter__(self) -> Any:
+        chunks = self._inner.__aiter__()
         try:
-            async for chunk in self._inner:
+            while True:
+                self._check_budget()
+                scope = asyncio.timeout(self._deadline.remaining())
+                try:
+                    async with scope:
+                        chunk = await anext(chunks)
+                except StopAsyncIteration:
+                    break
+                except Exception as exc:
+                    if scope.expired():
+                        raise self._expired() from exc
+                    raise
                 yield chunk
         except Exception as exc:
-            self._finish(Outcome(kind=FailureKind.BODY_ERROR, exception=exc))
+            self._failed(exc)
             raise
         self._finish(Outcome(kind=None))
 
@@ -378,10 +415,20 @@ async def aclose(self) -> None:
 
 class SyncTimedStreamMixin(TimedStreamCore):
     def __iter__(self) -> Any:
+        # A blocked read cannot be interrupted: the deadline is soft here and
+        # refuses to START a read once the budget is gone, so the overrun is at
+        # most one chunk or one read timeout.
+        chunks = iter(self._inner)
         try:
-            yield from self._inner
+            while True:
+                self._check_budget()
+                try:
+                    chunk = next(chunks)
+                except StopIteration:
+                    break
+                yield chunk
         except Exception as exc:
-            self._finish(Outcome(kind=FailureKind.BODY_ERROR, exception=exc))
+            self._failed(exc)
             raise
         self._finish(Outcome(kind=None))
 
@@ -403,12 +450,14 @@ def __init__(
         sdk: Any,
         request_view: Callable[[Any], Any],
         timed_stream: Callable[..., Any],
+        translate: Callable[[CallError], BaseException],
     ) -> None:
         self._default_timeout = default_timeout
         self._clock = clock
         self._sdk = sdk
         self._request_view = request_view
         self._timed_stream = timed_stream
+        self._translate = translate
 
     def wrap_request(self, native: Any) -> Any:
         return self._request_view(native)
@@ -440,11 +489,11 @@ async def discard(self, response: Any) -> None:
         except Exception:
             return None
 
-    def wrap_stream(self, response: Any, on_done: Callable[[Outcome, float], None]) -> None:
+    def wrap_stream(self, response: Any, on_done: Callable[[Outcome, float], None], deadline: Deadline) -> None:
         native = response.native
         stream = native.stream
         if isinstance(stream, self._sdk.AsyncByteStream):
-            native.stream = self._timed_stream(stream, self._clock, on_done)
+            native.stream = self._timed_stream(stream, self._clock, on_done, deadline, self._translate)
 
     def conn_metrics(self, response: Any) -> ConnMetrics | None:
         return None
@@ -461,12 +510,14 @@ def __init__(
         sdk: Any,
         request_view: Callable[[Any], Any],
         timed_stream: Callable[..., Any],
+        translate: Callable[[CallError], BaseException],
     ) -> None:
         self._default_timeout = default_timeout
         self._clock = clock
         self._sdk = sdk
         self._request_view = request_view
         self._timed_stream = timed_stream
+        self._translate = translate
 
     def wrap_request(self, native: Any) -> Any:
         return self._request_view(native)
@@ -498,11 +549,11 @@ def discard(self, response: Any) -> None:
         except Exception:
             return None
 
-    def wrap_stream(self, response: Any, on_done: Callable[[Outcome, float], None]) -> None:
+    def wrap_stream(self, response: Any, on_done: Callable[[Outcome, float], None], deadline: Deadline) -> None:
         native = response.native
         stream = native.stream
         if isinstance(stream, self._sdk.SyncByteStream):
-            native.stream = self._timed_stream(stream, self._clock, on_done)
+            native.stream = self._timed_stream(stream, self._clock, on_done, deadline, self._translate)
 
     def conn_metrics(self, response: Any) -> ConnMetrics | None:
         return None
@@ -636,7 +687,7 @@ def _compile(self, config: ClientConfig, native: Mapping[str, Mapping[str, Any]]
             applied.add(Capability.HTTP2)
         if config.proxy is not None:
             applied.add(Capability.PROXY)
-        emulated = {Capability.TIMEOUT_TOTAL}
+        emulated = {Capability.TIMEOUT_TOTAL, Capability.DEADLINE_COVERS_BODY}
         dropped: dict[Capability, str] = {}
         if sync:
             # The sync engine cannot cancel a blocked socket call: no hard

clientwright/adapters/aiohttp/capabilities.py

@@ -24,6 +24,7 @@
         Capability.TIMEOUT_WRITE: Support.ABSENT,
         Capability.TIMEOUT_POOL: Support.ABSENT,
         Capability.DEADLINE_HARD: Support.EMULATED,
+        Capability.DEADLINE_COVERS_BODY: Support.ABSENT,
         Capability.POOL_LIMIT_TOTAL: Support.NATIVE,
         Capability.POOL_LIMIT_PER_HOST: Support.NATIVE,
         Capability.KEEPALIVE: Support.NATIVE,
@@ -74,6 +75,10 @@
         ),
         "ceil_threshold": "ClientTimeout.ceil_threshold is raised so aiohttp never ceils deadlines to whole seconds.",
         "body_duration": "The middleware returns at headers; body read is not instrumented (no body_duration metric).",
+        "deadline_covers_body": (
+            "The body streams outside the seam through aiohttp's StreamReader: the total stops at the headers and "
+            "only sock_read bounds a dripping body."
+        ),
         "max_keepalive": "aiohttp has no cap on the NUMBER of keep-alive connections; pool.max_keepalive is ignored.",
     },
 )

clientwright/adapters/aiohttp/normalize.py

@@ -9,6 +9,7 @@
 from ...core.contracts.message import RequestView, ResponseView
 from ...core.engine.base import default_response_outcome
 from ...core.model import ConnMetrics, FailureKind, Outcome
+from ...core.policy.budget import Deadline
 from ._imports import aiohttp
 from .classify import classify_error
 from .trace import current_conn_metrics
@@ -52,9 +53,12 @@ async def discard(self, response: ResponseView) -> None:
         except Exception:
             return None
 
-    def wrap_stream(self, response: ResponseView, on_done: Callable[[Outcome, float], None]) -> None:
-        # The middleware returns at headers; the body streams outside the seam.
-        # Declared in capabilities (note "body_duration"), not silently skipped.
+    def wrap_stream(
+        self, response: ResponseView, on_done: Callable[[Outcome, float], None], deadline: Deadline
+    ) -> None:
+        # The middleware returns at headers; the body streams outside the seam
+        # through aiohttp's own StreamReader, so neither telemetry nor the
+        # deadline reaches it. Declared in capabilities, not silently skipped.
         return None
 
     def conn_metrics(self, response: ResponseView) -> ConnMetrics | None:

clientwright/adapters/httpx/normalize.py

@@ -6,6 +6,7 @@
 
 from .._httpx_shared import AsyncFamilyNormalizer, AsyncTimedStreamMixin
 from ._imports import httpx
+from .errors import translate_call_error
 from .views import HttpxRequestView
 
 
@@ -21,6 +22,7 @@ def __init__(self, default_timeout: dict[str, float | None], clock: Callable[[],
             sdk=httpx,
             request_view=lambda native: HttpxRequestView(native, default_timeout),
             timed_stream=_TimedAsyncStream,
+            translate=translate_call_error,
         )
 
 

clientwright/adapters/httpx/normalize_sync.py

@@ -6,6 +6,7 @@
 
 from .._httpx_shared import SyncFamilyNormalizer, SyncTimedStreamMixin
 from ._imports import httpx
+from .errors import translate_call_error
 from .views import HttpxRequestView
 
 
@@ -21,6 +22,7 @@ def __init__(self, default_timeout: dict[str, float | None], clock: Callable[[],
             sdk=httpx,
             request_view=lambda native: HttpxRequestView(native, default_timeout),
             timed_stream=_TimedSyncStream,
+            translate=translate_call_error,
         )
 
 

clientwright/adapters/httpx2/normalize.py

@@ -6,6 +6,7 @@
 
 from .._httpx_shared import AsyncFamilyNormalizer, AsyncTimedStreamMixin
 from ._imports import httpx2
+from .errors import translate_call_error
 from .views import HttpxRequestView
 
 
@@ -21,6 +22,7 @@ def __init__(self, default_timeout: dict[str, float | None], clock: Callable[[],
             sdk=httpx2,
             request_view=lambda native: HttpxRequestView(native, default_timeout),
             timed_stream=_TimedAsyncStream,
+            translate=translate_call_error,
         )
 
 

clientwright/adapters/httpx2/normalize_sync.py

@@ -6,6 +6,7 @@
 
 from .._httpx_shared import SyncFamilyNormalizer, SyncTimedStreamMixin
 from ._imports import httpx2
+from .errors import translate_call_error
 from .views import HttpxRequestView
 
 
@@ -21,6 +22,7 @@ def __init__(self, default_timeout: dict[str, float | None], clock: Callable[[],
             sdk=httpx2,
             request_view=lambda native: HttpxRequestView(native, default_timeout),
             timed_stream=_TimedSyncStream,
+            translate=translate_call_error,
         )
 
 

clientwright/adapters/requests/capabilities.py

@@ -24,6 +24,7 @@
         Capability.TIMEOUT_WRITE: Support.ABSENT,
         Capability.TIMEOUT_POOL: Support.ABSENT,
         Capability.DEADLINE_HARD: Support.ABSENT,
+        Capability.DEADLINE_COVERS_BODY: Support.ABSENT,
         Capability.POOL_LIMIT_TOTAL: Support.ABSENT,
         Capability.POOL_LIMIT_PER_HOST: Support.NATIVE,
         Capability.KEEPALIVE: Support.DEGRADED,
@@ -74,6 +75,10 @@
         "base_url": "requests has no base_url; the build rejects a config that sets one.",
         "per_call_options": "No request extensions; route/idempotency travel via the call_options() context manager.",
         "protocol_error": "requests folds protocol violations into ConnectionError; they surface as disconnected.",
+        "deadline_covers_body": (
+            "Session.send reads the body above the seam: the total stops at the headers and only the read timeout "
+            "bounds a dripping body."
+        ),
     },
 )
 

clientwright/adapters/requests/normalize.py

@@ -8,6 +8,7 @@
 from ...core.contracts.message import RequestView, ResponseView
 from ...core.engine.base import default_response_outcome
 from ...core.model import ConnMetrics, FailureKind, Outcome
+from ...core.policy.budget import Deadline
 from ._imports import requests
 from .classify import classify_error
 from .views import RequestsRequestView, RequestsResponseView
@@ -51,7 +52,9 @@ def discard(self, response: ResponseView) -> None:
         except Exception:
             return None
 
-    def wrap_stream(self, response: ResponseView, on_done: Callable[[Outcome, float], None]) -> None:
+    def wrap_stream(
+        self, response: ResponseView, on_done: Callable[[Outcome, float], None], deadline: Deadline
+    ) -> None:
         # Session.send consumes the body ABOVE this seam (unless stream=True);
         # body read is not instrumented. Declared in capabilities.
         return None

clientwright/adapters/urllib3/capabilities.py

@@ -24,6 +24,7 @@
         Capability.TIMEOUT_WRITE: Support.ABSENT,
         Capability.TIMEOUT_POOL: Support.NATIVE,
         Capability.DEADLINE_HARD: Support.ABSENT,
+        Capability.DEADLINE_COVERS_BODY: Support.ABSENT,
         Capability.POOL_LIMIT_TOTAL: Support.ABSENT,
         Capability.POOL_LIMIT_PER_HOST: Support.NATIVE,
         Capability.KEEPALIVE: Support.DEGRADED,
@@ -80,6 +81,10 @@
         "per_call_options": "No request object; route/idempotency travel via the call_options() context manager.",
         "base_url": "urllib3 has no base_url; the build rejects a config that sets one.",
         "proxy": "An explicit proxy builds a genuine urllib3.ProxyManager; env proxies are not read (dropped).",
+        "deadline_covers_body": (
+            "urlopen preloads or streams the body above the seam: the total stops at the headers and only the read "
+            "timeout bounds a dripping body."
+        ),
     },
 )
 

clientwright/adapters/urllib3/normalize.py

@@ -8,6 +8,7 @@
 from ...core.contracts.message import RequestView, ResponseView
 from ...core.engine.base import default_response_outcome
 from ...core.model import ConnMetrics, FailureKind, Outcome
+from ...core.policy.budget import Deadline
 from .classify import classify_error
 from .views import Urllib3RequestView, Urllib3ResponseView
 
@@ -39,7 +40,9 @@ def discard(self, response: ResponseView) -> None:
         except Exception:
             return None
 
-    def wrap_stream(self, response: ResponseView, on_done: Callable[[Outcome, float], None]) -> None:
+    def wrap_stream(
+        self, response: ResponseView, on_done: Callable[[Outcome, float], None], deadline: Deadline
+    ) -> None:
         # urlopen preloads the body by default; streaming reads happen above
         # the seam and are not instrumented. Declared in capabilities.
         return None

clientwright/core/capabilities.py

@@ -34,6 +34,7 @@ class Capability(StrEnum):
     TIMEOUT_WRITE = "timeout_write"
     TIMEOUT_POOL = "timeout_pool"
     DEADLINE_HARD = "deadline_hard"
+    DEADLINE_COVERS_BODY = "deadline_covers_body"
     POOL_LIMIT_TOTAL = "pool_limit_total"
     POOL_LIMIT_PER_HOST = "pool_limit_per_host"
     KEEPALIVE = "keepalive"