From 150444dfa5731d5038acdd38380779be728aa77d Mon Sep 17 00:00:00 2001
From: Alexey Shalaev <75322386+AlexeyShalaev@users.noreply.github.com>
Date: Mon, 7 Sep 2026 00:41:39 +0300
Subject: [PATCH] fix: bound the response body by the total on the httpx family

TimeoutConfig.total stopped at the response headers on httpx and httpx2:
the engine returns at handle_async_request, so a body that kept dripping
after the headers escaped the wall clock entirely and a 1 s total let a
4 s body through. The response stream already passes through the
family's timed-stream wrapper for the body-duration metric, so the
deadline now travels with it. The async wrapper bounds every chunk by
the remaining budget and raises the adapter's DeadlineExceededError when
the budget is gone; the sync wrapper cannot interrupt a blocked read and
instead refuses to start the next one, late by at most one chunk or one
read timeout. Both hand on_done an Outcome(kind=total_timeout).

The declaration is a new Capability.DEADLINE_COVERS_BODY rather than
boundary=FULL: DurationBoundary says where the call duration metric
closes, and that is still the headers. The httpx family declares it
emulated and reports it in the build; aiohttp, requests and urllib3
declare it absent with notes. The normalizer contract's wrap_stream
gains the deadline argument, and OriginServer gets a
/drip/{count}/{interval} route for the scenario.

Closes #25
---
 clientwright/adapters/_httpx_shared.py        |  77 ++++++++++--
 clientwright/adapters/aiohttp/capabilities.py |   5 +
 clientwright/adapters/aiohttp/normalize.py    |  10 +-
 clientwright/adapters/httpx/normalize.py      |   2 +
 clientwright/adapters/httpx/normalize_sync.py |   2 +
 clientwright/adapters/httpx2/normalize.py     |   2 +
 .../adapters/httpx2/normalize_sync.py         |   2 +
 .../adapters/requests/capabilities.py         |   5 +
 clientwright/adapters/requests/normalize.py   |   5 +-
 clientwright/adapters/urllib3/capabilities.py |   5 +
 clientwright/adapters/urllib3/normalize.py    |   5 +-
 clientwright/core/capabilities.py             |   1 +
 clientwright/core/contracts/message.py        |  13 +-
 clientwright/core/engine/aio.py               |  12 +-
 clientwright/core/engine/sync.py              |  12 +-
 clientwright/core/testing/origin.py           |  14 +++
 docs/adapters/aiohttp.md                      |  12 +-
 docs/adapters/httpx.md                        |   5 +-
 docs/agents.md                                |  17 ++-
 docs/guide/testing.md                         |   1 +
 docs/guide/timeouts.md                        |  12 ++
 docs/learn/sync-and-async.md                  |   3 +
 tests/helpers/engine.py                       |   5 +-
 tests/integration/adapters/test_aiohttp.py    |  13 ++
 .../integration/adapters/test_httpx_async.py  |  18 +++
 tests/integration/adapters/test_httpx_sync.py |  16 +++
 tests/integration/chaos/test_origin_faults.py |   9 ++
 tests/unit/adapters/test_httpx_shared.py      | 118 +++++++++++++++++-
 28 files changed, 352 insertions(+), 49 deletions(-)

diff --git a/clientwright/adapters/_httpx_shared.py b/clientwright/adapters/_httpx_shared.py
index 3fd968d..b34945d 100644
--- a/clientwright/adapters/_httpx_shared.py
+++ b/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
diff --git a/clientwright/adapters/aiohttp/capabilities.py b/clientwright/adapters/aiohttp/capabilities.py
index 6163f84..7a5265e 100644
--- a/clientwright/adapters/aiohttp/capabilities.py
+++ b/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.",
     },
 )
diff --git a/clientwright/adapters/aiohttp/normalize.py b/clientwright/adapters/aiohttp/normalize.py
index 9517c13..9ad50bc 100644
--- a/clientwright/adapters/aiohttp/normalize.py
+++ b/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:
diff --git a/clientwright/adapters/httpx/normalize.py b/clientwright/adapters/httpx/normalize.py
index 46b5e1c..5ccf55b 100644
--- a/clientwright/adapters/httpx/normalize.py
+++ b/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,
         )
 
 
diff --git a/clientwright/adapters/httpx/normalize_sync.py b/clientwright/adapters/httpx/normalize_sync.py
index 69c02eb..badbda7 100644
--- a/clientwright/adapters/httpx/normalize_sync.py
+++ b/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,
         )
 
 
diff --git a/clientwright/adapters/httpx2/normalize.py b/clientwright/adapters/httpx2/normalize.py
index 44fc708..4c854ed 100644
--- a/clientwright/adapters/httpx2/normalize.py
+++ b/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,
         )
 
 
diff --git a/clientwright/adapters/httpx2/normalize_sync.py b/clientwright/adapters/httpx2/normalize_sync.py
index 215c873..87a841d 100644
--- a/clientwright/adapters/httpx2/normalize_sync.py
+++ b/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,
         )
 
 
diff --git a/clientwright/adapters/requests/capabilities.py b/clientwright/adapters/requests/capabilities.py
index 48b11a4..36d6573 100644
--- a/clientwright/adapters/requests/capabilities.py
+++ b/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."
+        ),
     },
 )
 
diff --git a/clientwright/adapters/requests/normalize.py b/clientwright/adapters/requests/normalize.py
index 44382fc..cf63e12 100644
--- a/clientwright/adapters/requests/normalize.py
+++ b/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
diff --git a/clientwright/adapters/urllib3/capabilities.py b/clientwright/adapters/urllib3/capabilities.py
index 9d41b03..fecef61 100644
--- a/clientwright/adapters/urllib3/capabilities.py
+++ b/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."
+        ),
     },
 )
 
diff --git a/clientwright/adapters/urllib3/normalize.py b/clientwright/adapters/urllib3/normalize.py
index 6562efb..0b6f79a 100644
--- a/clientwright/adapters/urllib3/normalize.py
+++ b/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
diff --git a/clientwright/core/capabilities.py b/clientwright/core/capabilities.py
index 54b781f..bd3e854 100644
--- a/clientwright/core/capabilities.py
+++ b/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"
diff --git a/clientwright/core/contracts/message.py b/clientwright/core/contracts/message.py
index fda8894..ebcfbc7 100644
--- a/clientwright/core/contracts/message.py
+++ b/clientwright/core/contracts/message.py
@@ -12,6 +12,7 @@
 from typing import Any, Protocol, runtime_checkable
 
 from ..model import ConnMetrics, FailureKind, Outcome, RequestInfo, ResolvedTimeouts
+from ..policy.budget import Deadline
 
 
 @runtime_checkable
@@ -80,9 +81,11 @@ async def discard(self, response: ResponseView) -> None:
         """MANDATORY before a repeat - otherwise the connection never returns to the pool."""
         ...
 
-    def wrap_stream(self, response: ResponseView, on_done: Callable[[Outcome, float], None]) -> None:
-        """boundary=full: wrap the body stream so read duration and errors reach
-        telemetry. No-op where there is nothing to wrap."""
+    def wrap_stream(
+        self, response: ResponseView, on_done: Callable[[Outcome, float], None], deadline: Deadline
+    ) -> None:
+        """Wrap the body stream so read duration and errors reach telemetry and
+        the deadline bounds the body. No-op where the body streams outside the seam."""
         ...
 
     def conn_metrics(self, response: ResponseView) -> ConnMetrics | None: ...
@@ -105,7 +108,9 @@ def rewind(self, request: RequestView) -> None: ...
 
     def discard(self, response: ResponseView) -> 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: ...
 
     def conn_metrics(self, response: ResponseView) -> ConnMetrics | None: ...
 
diff --git a/clientwright/core/engine/aio.py b/clientwright/core/engine/aio.py
index e0b5104..0b029bd 100644
--- a/clientwright/core/engine/aio.py
+++ b/clientwright/core/engine/aio.py
@@ -58,7 +58,8 @@ async def run(self, native_request: Any, send: AsyncSend) -> Any:
         final_outcome = Outcome(kind=FailureKind.UNKNOWN)
         response: ResponseView | None = None
         try:
-            response, final_outcome = await self._admitted(request, send, observation)
+            deadline = self._deadline(request)
+            response, final_outcome = await self._admitted(request, send, observation, deadline)
         except CallError as error:
             final_outcome = self._call_error_outcome(error)
             raise self._translate(error) from error
@@ -71,7 +72,7 @@ async def run(self, native_request: Any, send: AsyncSend) -> Any:
         if final_outcome.exception is not None:
             raise final_outcome.exception
         assert response is not None  # a call without exception always has a response
-        self._wrap_stream(response, info)
+        self._wrap_stream(response, info, deadline)
         return response.native
 
     def _call_error_outcome(self, error: CallError) -> Outcome:
@@ -81,11 +82,11 @@ def _call_error_outcome(self, error: CallError) -> Outcome:
             return Outcome(kind=FailureKind.TOTAL_TIMEOUT, exception=error)
         return Outcome(kind=FailureKind.UNKNOWN, exception=error)
 
-    def _wrap_stream(self, response: ResponseView, info: Any) -> None:
+    def _wrap_stream(self, response: ResponseView, info: Any, deadline: Deadline) -> None:
         def on_done(outcome: Outcome, duration: float) -> None:
             self._telemetry.record_body_duration(info, duration)
 
-        self._norm.wrap_stream(response, on_done)
+        self._norm.wrap_stream(response, on_done, deadline)
 
     def _inject_headers(self, request: RequestView) -> None:
         headers = request.headers
@@ -103,12 +104,11 @@ def _deadline(self, request: RequestView) -> Deadline:
         return Deadline.intersect(self._runtime.clock, self._plan.config.timeout.total, ambient)
 
     async def _admitted(
-        self, request: RequestView, send: AsyncSend, observation: CallObservation
+        self, request: RequestView, send: AsyncSend, observation: CallObservation, deadline: Deadline
     ) -> tuple[ResponseView | None, Outcome]:
         plan = self._plan
         runtime = self._runtime
         info = request.info
-        deadline = self._deadline(request)
         self._inject_headers(request)
         circuit_key: str | None = None
         async with AsyncExitStack() as stack:
diff --git a/clientwright/core/engine/sync.py b/clientwright/core/engine/sync.py
index fa7a6d6..986bae1 100644
--- a/clientwright/core/engine/sync.py
+++ b/clientwright/core/engine/sync.py
@@ -57,7 +57,8 @@ def run(self, native_request: Any, send: SyncSend) -> Any:
         final_outcome = Outcome(kind=FailureKind.UNKNOWN)
         response: ResponseView | None = None
         try:
-            response, final_outcome = self._admitted(request, send, observation)
+            deadline = self._deadline()
+            response, final_outcome = self._admitted(request, send, observation, deadline)
         except CallError as error:
             final_outcome = self._call_error_outcome(error)
             raise self._translate(error) from error
@@ -70,7 +71,7 @@ def run(self, native_request: Any, send: SyncSend) -> Any:
         if final_outcome.exception is not None:
             raise final_outcome.exception
         assert response is not None
-        self._wrap_stream(response, info)
+        self._wrap_stream(response, info, deadline)
         return response.native
 
     def _call_error_outcome(self, error: CallError) -> Outcome:
@@ -80,11 +81,11 @@ def _call_error_outcome(self, error: CallError) -> Outcome:
             return Outcome(kind=FailureKind.TOTAL_TIMEOUT, exception=error)
         return Outcome(kind=FailureKind.UNKNOWN, exception=error)
 
-    def _wrap_stream(self, response: ResponseView, info: Any) -> None:
+    def _wrap_stream(self, response: ResponseView, info: Any, deadline: Deadline) -> None:
         def on_done(outcome: Outcome, duration: float) -> None:
             self._telemetry.record_body_duration(info, duration)
 
-        self._norm.wrap_stream(response, on_done)
+        self._norm.wrap_stream(response, on_done, deadline)
 
     def _inject_headers(self, request: RequestView) -> None:
         headers = request.headers
@@ -102,12 +103,11 @@ def _deadline(self) -> Deadline:
         return Deadline.intersect(self._runtime.clock, self._plan.config.timeout.total, ambient)
 
     def _admitted(
-        self, request: RequestView, send: SyncSend, observation: CallObservation
+        self, request: RequestView, send: SyncSend, observation: CallObservation, deadline: Deadline
     ) -> tuple[ResponseView | None, Outcome]:
         plan = self._plan
         runtime = self._runtime
         info = request.info
-        deadline = self._deadline()
         self._inject_headers(request)
         circuit_key: str | None = None
         with ExitStack() as stack:
diff --git a/clientwright/core/testing/origin.py b/clientwright/core/testing/origin.py
index 36e2122..7fa311f 100644
--- a/clientwright/core/testing/origin.py
+++ b/clientwright/core/testing/origin.py
@@ -12,6 +12,7 @@
 
 Chaos routes (mid-stream and protocol-level faults):
 - ``/hang-body/{seconds}``           200 announcing 10 bytes: 3 arrive, the rest after the stall
+- ``/drip/{count}/{interval}``       200 announcing {count} bytes, one arriving every {interval} seconds
 - ``/drop-body``                     200 announcing 10 bytes but the connection dies after 3
 - ``/garbage``                       raw non-HTTP bytes instead of a status line
 - ``/reset``                         hard TCP reset (SO_LINGER 0) instead of a response
@@ -108,6 +109,19 @@ def _handle(self) -> None:
             except OSError:  # the client gave up mid-stall; routine for this route
                 self.close_connection = True
             return
+        if parts[0] == "drip":
+            count, interval = int(parts[1]), float(parts[2])
+            self.send_response(200)
+            self.send_header("Content-Length", str(count))
+            self.end_headers()
+            try:
+                for _ in range(count):
+                    time.sleep(interval)
+                    self.wfile.write(b"x")
+                    self.wfile.flush()
+            except OSError:  # the client gave up mid-body; routine for this route
+                self.close_connection = True
+            return
         if parts[0] == "drop-body":
             self.send_response(200)
             self.send_header("Content-Length", "10")
diff --git a/docs/adapters/aiohttp.md b/docs/adapters/aiohttp.md
index e96720f..6e36fda 100644
--- a/docs/adapters/aiohttp.md
+++ b/docs/adapters/aiohttp.md
@@ -68,11 +68,13 @@ not as chores:
 ## Capability notes
 
 - **Boundary: headers.** The aiohttp seam completes when response *headers*
-  arrive; the body is read by your code afterwards. Call metrics therefore time
-  to-headers, with body read time reported separately as
-  `http_client_body_duration_seconds` (measured via the trace hooks). A body
-  that fails mid-read after a `200` is an error your code sees, but the call
-  metric has honestly already closed — this is declared, not hidden.
+  arrive; the body is read by your code afterwards through aiohttp's own
+  `StreamReader`. Call metrics therefore time to-headers, there is no
+  `http_client_body_duration_seconds`, and `timeout.total` stops there as well —
+  a body that keeps dripping is bounded only by `read` (`sock_read`). A body that
+  fails mid-read after a `200` is an error your code sees, but the call metric
+  has honestly already closed — this is declared (`deadline_covers_body: absent`),
+  not hidden.
 - **DNS errors**: natively distinguishable (`dns_error` is real here, unlike
   httpx).
 - **Pool wait**: folded by aiohttp into the connect phase — `pool_timeout` is
diff --git a/docs/adapters/httpx.md b/docs/adapters/httpx.md
index d3efe2c..5326eb6 100644
--- a/docs/adapters/httpx.md
+++ b/docs/adapters/httpx.md
@@ -43,7 +43,10 @@ await client.post(
 ## Capability notes
 
 - **Total deadline**: hard on async (cancellation), soft on sync — httpx itself
-  has *no* wall-clock timeout; this is the engine's addition.
+  has *no* wall-clock timeout; this is the engine's addition. It covers the
+  response body too, because the response stream passes through the adapter:
+  the async client bounds every chunk by the remaining budget, the sync client
+  refuses the next chunk once the budget is gone.
 - **Errors**: httpx's exception taxonomy is the richest of the five, so
   `FailureKind` mapping is nearly one-to-one (`ConnectTimeout` →
   `connect_timeout`, `ReadTimeout` → `read_timeout`, ...). DNS failures are not
diff --git a/docs/agents.md b/docs/agents.md
index 404baa2..5d40b0c 100644
--- a/docs/agents.md
+++ b/docs/agents.md
@@ -295,7 +295,7 @@ exports.
 | `CallerOverride` | `CALLER_WINS`, `CONFIG_WINS`, `RAISE` |
 | `UnsupportedPolicy` | `IGNORE`, `WARN`, `STRICT` |
 | `Support` | `NATIVE`, `EMULATED`, `DEGRADED`, `ABSENT` |
-| `Capability` | `timeout_total`, `timeout_attempt`, `timeout_connect`, `timeout_read`, `timeout_write`, `timeout_pool`, `deadline_hard`, `pool_limit_total`, `pool_limit_per_host`, `keepalive`, `pool_metrics`, `conn_metrics`, `redirects_ownable`, `native_retry_disableable`, `per_call_options`, `retrofit`, `exact_native_type`, `balancer`, `http2`, `http3`, `proxy` |
+| `Capability` | `timeout_total`, `timeout_attempt`, `timeout_connect`, `timeout_read`, `timeout_write`, `timeout_pool`, `deadline_hard`, `deadline_covers_body`, `pool_limit_total`, `pool_limit_per_host`, `keepalive`, `pool_metrics`, `conn_metrics`, `redirects_ownable`, `native_retry_disableable`, `per_call_options`, `retrofit`, `exact_native_type`, `balancer`, `http2`, `http3`, `proxy` |
 | `SeamGranularity` | `HOP`, `LOGICAL` |
 | `DurationBoundary` | `HEADERS`, `FULL` |
 
@@ -351,7 +351,7 @@ task-local and thread-local, inherited by tasks started inside it, invisible to
 
 `OriginServer` routes: `/echo`, `/status/{code}`, `/slow/{seconds}`, `/redirect/{n}`,
 `/redirect-loop`, `/flaky/{key}/{fails}`, `/retry-after/{seconds}`, `/disconnect`,
-`/hang-body/{seconds}`, `/drop-body`, `/garbage`, `/reset`,
+`/hang-body/{seconds}`, `/drip/{count}/{interval}`, `/drop-body`, `/garbage`, `/reset`,
 `/flaky-disconnect/{key}/{fails}`. It carries `.url` and `.request_count(prefix)`.
 
 ### Telemetry
@@ -404,12 +404,14 @@ What each one will not do:
   implementation. Per-host pool limits are *emulated* by a per-origin in-flight semaphore
   (it limits requests, not connections). DNS failures collapse into `connect_error`.
   Everything TLS- and pool-related goes into the transport constructor, because
-  `Client(transport=...)` silently ignores `verify`, `http2` and `limits`.
+  `Client(transport=...)` silently ignores `verify`, `http2` and `limits`. The response
+  stream passes through the adapter, so the total bounds the body too.
 * **aiohttp** — the session must be constructed inside a running event loop. Write and pool
   timeouts do not exist and are dropped; `pool.max_keepalive` is ignored; HTTP/2 is dropped.
   `ClientTimeout.total` is deliberately `None` so aiohttp's timer cannot wrap the engine's
-  own retry loop. The call duration ends at the headers, so there is no body-duration
-  metric. A caller writing `session.get(url, middlewares=())` replaces the chain and
+  own retry loop. The call duration and the total end at the headers, so there is no
+  body-duration metric and a dripping body is bounded only by `read`. A caller writing
+  `session.get(url, middlewares=())` replaces the chain and
   bypasses the engine entirely — that is counted, not prevented.
 * **requests** — no `base_url` (a config that sets one fails the build), no write or pool
   timeout, no `attempt` ceiling, no hard deadline. It closes requests' famous hole: the
@@ -464,7 +466,10 @@ What each one will not do:
 10. **The total deadline covers everything and is only hard on async.** Async engines wrap
     each attempt in a cancellation scope; sync engines cannot cancel a blocked socket, so
     they clamp phases and re-check at attempt boundaries — the failure then arrives as
-    `read_timeout`, not `total_timeout`. Sync adapters declare `deadline_hard: absent`.
+    `read_timeout`, not `total_timeout`. Sync adapters declare `deadline_hard: absent`. The
+    total reaches into the response body only on the httpx family
+    (`deadline_covers_body: emulated` — cancellation per chunk on async, a check between
+    chunks on sync); on aiohttp, requests and urllib3 it stops at the headers.
 11. **`caller_override` never lets a caller escape the total.** `CALLER_WINS` replaces the
     config's phases with the caller's, then clamps them to the remaining budget; a
     `timeout=60` on a call with 3 seconds left gets 3 seconds. `RAISE` makes a per-call
diff --git a/docs/guide/testing.md b/docs/guide/testing.md
index 19b2c8e..691bfb8 100644
--- a/docs/guide/testing.md
+++ b/docs/guide/testing.md
@@ -60,6 +60,7 @@ The route table reads like a chaos menu:
 | `/disconnect` | closes without a response |
 | `/drop-body` | announces 10 body bytes, dies after 3 |
 | `/hang-body/{seconds}` | sends 3 bytes, stalls, then finishes |
+| `/drip/{count}/{interval}` | announces *count* bytes and sends one every *interval* seconds |
 | `/garbage` | raw non-HTTP bytes instead of a status line |
 | `/reset` | a hard TCP reset |
 
diff --git a/docs/guide/timeouts.md b/docs/guide/timeouts.md
index 39a922c..6d28711 100644
--- a/docs/guide/timeouts.md
+++ b/docs/guide/timeouts.md
@@ -34,6 +34,18 @@ scope, so a stuck read is cancelled mid-flight and the call raises with the outc
 [Sync and async](../learn/sync-and-async.md#the-one-honest-difference-hard-vs-soft-deadlines)
 for exactly what that means and why it is declared rather than hidden.
 
+On the httpx family the total reaches into the response body as well: the adapter
+wraps the response stream, so a body that keeps dripping after the headers is cut
+when the budget runs out — by cancellation on the async client, and on the sync
+client by refusing the next chunk once the budget is gone (a blocked read still
+cannot be interrupted, so the sync overrun is at most one chunk or one read
+timeout). The call raises `DeadlineExceededError` either way. The call metric had
+already closed at the headers, so a body-phase deadline shows in the exception and
+in `http_client_body_duration_seconds`, not in `requests_total`. On aiohttp,
+requests and urllib3 the seam ends at the headers and only the SDK's read timeout
+bounds the body; `Capability.DEADLINE_COVERS_BODY` in the
+[capability record](capabilities.md) says which is which.
+
 ## Phase timeouts
 
 `connect`, `read`, `write` and `pool_acquire` cap phases of a *single attempt* and
diff --git a/docs/learn/sync-and-async.md b/docs/learn/sync-and-async.md
index 7e0e59b..e968a8d 100644
--- a/docs/learn/sync-and-async.md
+++ b/docs/learn/sync-and-async.md
@@ -41,6 +41,9 @@ enforces the total as a **soft** deadline: it clamps every phase timeout of ever
 attempt to the remaining budget and re-checks the wall clock at attempt boundaries.
 You still never wait meaningfully longer than `total` — but the failure arrives as
 the clamped phase (`read_timeout` from the SDK), not as an abstract deadline error.
+The response body follows the same split on the httpx family: the async client
+cancels a dripping body the moment the total runs out, the sync client refuses the
+next chunk after it.
 
 This is deliberately *not* papered over. Sync adapters declare
 `DEADLINE_HARD: absent` in their [capability record](../guide/capabilities.md), and
diff --git a/tests/helpers/engine.py b/tests/helpers/engine.py
index 7c3fcac..abd11c4 100644
--- a/tests/helpers/engine.py
+++ b/tests/helpers/engine.py
@@ -28,6 +28,7 @@
 from clientwright.core.engine.sync import SyncAttemptEngine
 from clientwright.core.model import ConnMetrics, FailureKind, Outcome, RequestInfo, ResolvedTimeouts, origin_of
 from clientwright.core.plan import ClientRuntime, compile_plan
+from clientwright.core.policy.budget import Deadline
 from clientwright.core.telemetry.emitter import ClientTelemetry
 from clientwright.core.testing import RecordingMetrics
 from tests.helpers.views import FakeResponse
@@ -211,7 +212,9 @@ def classify_error(self, exc: BaseException) -> FailureKind:
     def classify_response(self, response: FakeResponse) -> Outcome:
         return default_response_outcome(response)
 
-    def wrap_stream(self, response: FakeResponse, on_done: Callable[[Outcome, float], None]) -> None:
+    def wrap_stream(
+        self, response: FakeResponse, on_done: Callable[[Outcome, float], None], deadline: Deadline
+    ) -> None:
         self.wrapped_streams += 1
 
     def conn_metrics(self, response: FakeResponse) -> ConnMetrics | None:
diff --git a/tests/integration/adapters/test_aiohttp.py b/tests/integration/adapters/test_aiohttp.py
index d1353b7..acb18d9 100644
--- a/tests/integration/adapters/test_aiohttp.py
+++ b/tests/integration/adapters/test_aiohttp.py
@@ -15,10 +15,12 @@
 import clientwright  # noqa: E402
 from clientwright import AdapterDeps, ClientConfig, RetryConfig, TimeoutConfig  # noqa: E402
 from clientwright.adapters.aiohttp import (  # noqa: E402
+    CAPABILITIES,
     AiohttpCircuitOpenError,
     AiohttpTooManyRedirectsError,
     call_options,
 )
+from clientwright.core.capabilities import Capability, Support  # noqa: E402
 from clientwright.core.config import CircuitBreakerConfig  # noqa: E402
 from clientwright.core.testing import OriginServer, RecordingMetrics  # noqa: E402
 from tests.helpers.telemetry import RecordingTracer  # noqa: E402
@@ -185,6 +187,17 @@ async def test__deadline_header__stamped_with_remaining_budget(origin: OriginSer
     await client.close()
 
 
+async def test__dripping_body__outside_the_seam_by_declaration(origin: OriginServer, deps: AdapterDeps) -> None:
+    # The middleware returns at the headers and the body is aiohttp's own
+    # StreamReader: the total cannot reach it, and the record says so.
+    assert CAPABILITIES.support_of(Capability.DEADLINE_COVERS_BODY) is Support.ABSENT
+    config = base_config(origin, timeout=TimeoutConfig(total=0.3, connect=1.0), retry=None)
+    client = await build(config, deps)
+    response = await client.get("/drip/3/0.2")
+    assert await response.read() == b"xxx"  # 0.6 s of body arrives after a 0.3 s total
+    await client.close()
+
+
 # --- circuit breaker ---
 
 
diff --git a/tests/integration/adapters/test_httpx_async.py b/tests/integration/adapters/test_httpx_async.py
index b8c12b6..558af0f 100644
--- a/tests/integration/adapters/test_httpx_async.py
+++ b/tests/integration/adapters/test_httpx_async.py
@@ -14,6 +14,7 @@
     IDEMPOTENT_EXTENSION,
     ROUTE_EXTENSION,
     HttpxCircuitOpenError,
+    HttpxDeadlineExceededError,
     HttpxTooManyRedirectsError,
 )
 from clientwright.core.config import CircuitBreakerConfig  # noqa: E402
@@ -175,6 +176,23 @@ async def test__deadline_header__stamped_with_remaining_budget(origin: OriginSer
     await client.aclose()
 
 
+async def test__dripping_body__cut_by_the_total(
+    origin: OriginServer, metrics: RecordingMetrics, deps: AdapterDeps
+) -> None:
+    config = base_config(origin, timeout=TimeoutConfig(total=0.5, connect=1.0), retry=None)
+    client = await build(config, deps)
+    started = asyncio.get_running_loop().time()
+    with pytest.raises(HttpxDeadlineExceededError) as excinfo:
+        await client.get("/drip/6/0.2")  # 1.2 s of body behind instant headers
+    elapsed = asyncio.get_running_loop().time() - started
+    assert isinstance(excinfo.value, httpx.TimeoutException)
+    assert 0.4 < elapsed < 1.0
+    assert metrics.calls[0]["outcome"] == "success"  # the call metric closed at the headers, as declared
+    assert len(metrics.body_durations) == 1
+    assert metrics.inflight_balance == 0
+    await client.aclose()
+
+
 # --- circuit breaker ---
 
 
diff --git a/tests/integration/adapters/test_httpx_sync.py b/tests/integration/adapters/test_httpx_sync.py
index ecb3e62..d313cc0 100644
--- a/tests/integration/adapters/test_httpx_sync.py
+++ b/tests/integration/adapters/test_httpx_sync.py
@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import time
+
 import pytest
 
 httpx = pytest.importorskip("httpx", reason="requires the [httpx] extra")
@@ -10,6 +12,7 @@
 from clientwright import AdapterDeps, ClientConfig, RetryConfig, TimeoutConfig  # noqa: E402
 from clientwright.adapters.httpx import (  # noqa: E402
     HttpxCircuitOpenError,
+    HttpxDeadlineExceededError,
     HttpxTooManyRedirectsError,
 )
 from clientwright.core.config import CircuitBreakerConfig  # noqa: E402
@@ -68,6 +71,19 @@ def test__soft_deadline__clamps_read_phase(origin: OriginServer, deps: AdapterDe
     client.close()
 
 
+def test__dripping_body__refused_after_the_total(
+    origin: OriginServer, metrics: RecordingMetrics, deps: AdapterDeps
+) -> None:
+    config = base_config(origin, timeout=TimeoutConfig(total=0.5, connect=1.0), retry=None)
+    client = build(config, deps)
+    started = time.monotonic()
+    with pytest.raises(HttpxDeadlineExceededError):
+        client.get("/drip/6/0.2")  # 1.2 s of body behind instant headers
+    assert time.monotonic() - started < 1.0  # soft: late by at most the chunk already in flight
+    assert metrics.calls[0]["outcome"] == "success"  # the call metric closed at the headers, as declared
+    client.close()
+
+
 def test__circuit_breaker__opens_after_threshold(
     origin: OriginServer, metrics: RecordingMetrics, deps: AdapterDeps
 ) -> None:
diff --git a/tests/integration/chaos/test_origin_faults.py b/tests/integration/chaos/test_origin_faults.py
index 0c55b6e..2905f19 100644
--- a/tests/integration/chaos/test_origin_faults.py
+++ b/tests/integration/chaos/test_origin_faults.py
@@ -8,6 +8,7 @@
 from __future__ import annotations
 
 import http.client
+import time
 from urllib.parse import urlsplit
 
 import pytest
@@ -63,6 +64,14 @@ def test__hang_body__delivers_everything_to_a_patient_reader(origin: OriginServe
     assert response.read() == b"0123456789"
 
 
+def test__drip__delivers_one_byte_per_interval(origin: OriginServer) -> None:
+    started = time.monotonic()
+    response = _get(origin, "/drip/3/0.05")
+    assert response.getheader("Content-Length") == "3"
+    assert response.read() == b"xxx"
+    assert time.monotonic() - started >= 0.15
+
+
 def test__flaky_disconnect__drops_then_recovers(origin: OriginServer) -> None:
     first = _connection(origin)
     first.request("GET", "/flaky-disconnect/selftest/1")
diff --git a/tests/unit/adapters/test_httpx_shared.py b/tests/unit/adapters/test_httpx_shared.py
index 28111a0..50609d8 100644
--- a/tests/unit/adapters/test_httpx_shared.py
+++ b/tests/unit/adapters/test_httpx_shared.py
@@ -11,6 +11,7 @@
 
 import asyncio
 import ssl
+import time
 from collections.abc import AsyncIterator, Iterator
 from types import SimpleNamespace
 
@@ -26,7 +27,9 @@
     no_proxy_hosts,
     ssl_arguments,
 )
+from clientwright.adapters.httpx import HttpxDeadlineExceededError  # noqa: E402
 from clientwright.adapters.httpx.classify import classify_error  # noqa: E402
+from clientwright.adapters.httpx.errors import translate_call_error  # noqa: E402
 from clientwright.adapters.httpx.normalize import AsyncHttpxNormalizer  # noqa: E402
 from clientwright.adapters.httpx.normalize_sync import SyncHttpxNormalizer  # noqa: E402
 from clientwright.adapters.httpx.transport import (  # noqa: E402
@@ -38,6 +41,7 @@
 from clientwright.core.capabilities import Capability  # noqa: E402
 from clientwright.core.config import ClientConfig, PoolConfig, ProxyConfig, TlsConfig  # noqa: E402
 from clientwright.core.model import FailureKind, Outcome  # noqa: E402
+from clientwright.core.policy.budget import Deadline  # noqa: E402
 
 DEFAULT_TIMEOUT = {"connect": 5.0, "read": 5.0, "write": 5.0, "pool": 5.0}
 
@@ -149,6 +153,39 @@ def __iter__(self) -> Iterator[bytes]:
         raise ValueError("mid-body failure")
 
 
+class _DrippingAsyncStream(httpx.AsyncByteStream):
+    """One chunk at once, then a stall longer than any deadline in this file."""
+
+    async def __aiter__(self) -> AsyncIterator[bytes]:
+        yield b"first"
+        await asyncio.sleep(5.0)
+        yield b"late"
+
+
+class _CountingAsyncStream(httpx.AsyncByteStream):
+    def __init__(self) -> None:
+        self.reads = 0
+
+    async def __aiter__(self) -> AsyncIterator[bytes]:
+        self.reads += 1
+        yield b"body"
+
+
+class _CountingSyncStream(httpx.SyncByteStream):
+    """Every chunk moves the fake clock by ``advance``: the time the read took."""
+
+    def __init__(self, clock: _FakeClock, advance: float) -> None:
+        self._clock = clock
+        self._advance = advance
+        self.reads = 0
+
+    def __iter__(self) -> Iterator[bytes]:
+        for _ in range(3):
+            self.reads += 1
+            self._clock.now += self._advance
+            yield b"x"
+
+
 class _OnDoneRecorder:
     def __init__(self) -> None:
         self.calls: list[tuple[Outcome, float]] = []
@@ -160,7 +197,7 @@ def __call__(self, outcome: Outcome, duration: float) -> None:
 async def test__async_timed_stream__reports_read_failure_once_and_reraises() -> None:
     recorder = _OnDoneRecorder()
     clock = _FakeClock(now=10.0)
-    stream = _TimedAsyncStream(_FailingAsyncStream(), clock, recorder)
+    stream = _TimedAsyncStream(_FailingAsyncStream(), clock, recorder, Deadline(None, clock), translate_call_error)
     clock.now = 10.25
     received: list[bytes] = []
     with pytest.raises(ValueError, match="mid-body failure"):
@@ -178,7 +215,7 @@ async def test__async_timed_stream__reports_read_failure_once_and_reraises() ->
 def test__sync_timed_stream__reports_read_failure_once_and_reraises() -> None:
     recorder = _OnDoneRecorder()
     clock = _FakeClock(now=10.0)
-    stream = _TimedSyncStream(_FailingSyncStream(), clock, recorder)
+    stream = _TimedSyncStream(_FailingSyncStream(), clock, recorder, Deadline(None, clock), translate_call_error)
     clock.now = 10.25
     received: list[bytes] = []
     with pytest.raises(ValueError, match="mid-body failure"):
@@ -192,6 +229,74 @@ def test__sync_timed_stream__reports_read_failure_once_and_reraises() -> None:
     assert len(recorder.calls) == 1
 
 
+async def test__async_timed_stream__cuts_a_dripping_body_at_the_deadline() -> None:
+    recorder = _OnDoneRecorder()
+    deadline = Deadline(0.05, time.monotonic)
+    stream = _TimedAsyncStream(_DrippingAsyncStream(), time.monotonic, recorder, deadline, translate_call_error)
+    received: list[bytes] = []
+    with pytest.raises(HttpxDeadlineExceededError) as excinfo:
+        async for chunk in stream:
+            received.append(chunk)
+    assert isinstance(excinfo.value, httpx.TimeoutException)
+    assert received == [b"first"]
+    outcome, _ = recorder.calls[0]
+    assert outcome.kind is FailureKind.TOTAL_TIMEOUT
+    await stream.aclose()  # a later close must NOT report a second outcome
+    assert len(recorder.calls) == 1
+
+
+async def test__async_timed_stream__refuses_to_read_once_the_budget_is_gone() -> None:
+    clock = _FakeClock(now=10.0)
+    deadline = Deadline(1.0, clock)
+    clock.now = 12.0  # the budget went while the caller sat on the response
+    inner = _CountingAsyncStream()
+    recorder = _OnDoneRecorder()
+    stream = _TimedAsyncStream(inner, clock, recorder, deadline, translate_call_error)
+    with pytest.raises(HttpxDeadlineExceededError):
+        async for _ in stream:
+            pass
+    assert inner.reads == 0
+    assert recorder.calls[0][0].kind is FailureKind.TOTAL_TIMEOUT
+
+
+async def test__async_timed_stream__body_inside_the_budget_completes() -> None:
+    clock = _FakeClock(now=10.0)
+    recorder = _OnDoneRecorder()
+    stream = _TimedAsyncStream(_CountingAsyncStream(), clock, recorder, Deadline(1.0, clock), translate_call_error)
+    assert [chunk async for chunk in stream] == [b"body"]
+    assert recorder.calls[0][0].kind is None
+
+
+def test__sync_timed_stream__refuses_the_next_read_once_the_deadline_passed() -> None:
+    clock = _FakeClock(now=10.0)
+    deadline = Deadline(1.0, clock)
+    inner = _CountingSyncStream(clock, advance=0.6)
+    recorder = _OnDoneRecorder()
+    stream = _TimedSyncStream(inner, clock, recorder, deadline, translate_call_error)
+    received: list[bytes] = []
+    with pytest.raises(HttpxDeadlineExceededError):
+        received.extend(stream)
+    # The second read started with budget left and its chunk is delivered; the
+    # third is refused. Soft: late by at most the read that was already in flight.
+    assert received == [b"x", b"x"]
+    assert inner.reads == 2
+    outcome, _ = recorder.calls[0]
+    assert outcome.kind is FailureKind.TOTAL_TIMEOUT
+    stream.close()  # a later close must NOT report a second outcome
+    assert len(recorder.calls) == 1
+
+
+def test__sync_timed_stream__body_inside_the_budget_completes() -> None:
+    clock = _FakeClock(now=10.0)
+    recorder = _OnDoneRecorder()
+    inner = _CountingSyncStream(clock, advance=0.1)
+    stream = _TimedSyncStream(inner, clock, recorder, Deadline(1.0, clock), translate_call_error)
+    assert list(stream) == [b"x", b"x", b"x"]
+    outcome, duration = recorder.calls[0]
+    assert outcome.kind is None
+    assert duration == pytest.approx(0.3)
+
+
 # --- normalizer edge paths ---------------------------------------------------
 
 
@@ -352,6 +457,15 @@ def test__explicit_proxy__reported_applied_natively() -> None:
         handle.close()
 
 
+def test__deadline_covers_body__reported_emulated() -> None:
+    handle = clientwright.build_sync_handle("httpx", ClientConfig(service_name="s"))
+    try:
+        assert Capability.DEADLINE_COVERS_BODY in handle.report.emulated
+    finally:
+        assert handle.close is not None
+        handle.close()
+
+
 def test__per_host_limit__reported_emulated_not_dropped() -> None:
     config = ClientConfig(service_name="s", pool=PoolConfig(max_connections_per_host=5))
     handle = clientwright.build_sync_handle("httpx", config)