feat(resilience): Retry refuses requests with streaming bodies

Closes the deferred-work item 'Retry + streaming bodies'. When the
request was constructed with an async-iterable content/data/files,
_request_with_body marked request.extensions['httpware.streaming_body']
= True. Retry now reads the marker and re-raises the original failure
with a PEP-678 note ('not retrying — request body is a stream that
cannot replay across attempts') instead of retrying with a consumed
iterator.

Check happens BEFORE budget.try_withdraw() so a refused retry doesn't
consume a budget token.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: lesnik512

src/httpware/middleware/resilience/retry.py

@@ -90,7 +90,7 @@ def __init__(  # noqa: PLR0913 — retry policy has many orthogonal knobs; a dat
         self.budget = budget if budget is not None else RetryBudget()
         self._sleep = _sleep
 
-    async def __call__(self, request: httpx2.Request, next: Next) -> httpx2.Response:  # noqa: A002, C901, PLR0912 — complexity budget: 3 error clauses + idempotency gate + budget gate + Retry-After branch + backoff
+    async def __call__(self, request: httpx2.Request, next: Next) -> httpx2.Response:  # noqa: A002, C901, PLR0912, PLR0915 — complexity budget: 3 error clauses + idempotency gate + streaming-body refusal + budget gate + Retry-After branch + backoff
         """Process a request through the retry loop. See module docstring."""
         method_eligible = request.method.upper() in self.retry_methods
         last_exc: BaseException | None = None
@@ -106,24 +106,46 @@ async def __call__(self, request: httpx2.Request, next: Next) -> httpx2.Response
                 else:
                     return await next(request)
             except StatusError as exc:
-                if not method_eligible or exc.response.status_code not in self.retry_status_codes:
+                retryable_status = exc.response.status_code in self.retry_status_codes
+                if not method_eligible or not retryable_status:
+                    if retryable_status and request.extensions.get("httpware.streaming_body"):
+                        exc.add_note(
+                            "httpware: not retrying — request body is a stream that cannot replay across attempts"
+                        )
                     raise
                 last_exc = exc
                 last_response = exc.response
             except (NetworkError, TimeoutError) as exc:
                 if not method_eligible:
+                    if request.extensions.get("httpware.streaming_body"):
+                        exc.add_note(
+                            "httpware: not retrying — request body is a stream that cannot replay across attempts"
+                        )
                     raise
                 last_exc = exc
                 last_response = None
             except builtins.TimeoutError as exc:
                 wrapped = TimeoutError("attempt timed out")
                 wrapped.__cause__ = exc  # set now; the retry path (last_exc = wrapped) has no `from` clause
                 if not method_eligible:
+                    if request.extensions.get("httpware.streaming_body"):
+                        wrapped.add_note(
+                            "httpware: not retrying — request body is a stream that cannot replay across attempts"
+                        )
                     raise wrapped from exc
                 last_exc = wrapped
                 last_response = None
 
             # ---- retryable failure path
+            if request.extensions.get("httpware.streaming_body"):
+                if last_exc is None:  # pragma: no cover — invariant from except branch
+                    msg = "Retry: streaming-body refusal reached with no last_exc"
+                    raise AssertionError(msg)
+                last_exc.add_note(
+                    "httpware: not retrying — request body is a stream that cannot replay across attempts"
+                )
+                raise last_exc
+
             if is_last:
                 if last_exc is None:  # pragma: no cover — structural invariant from except branch
                     msg = "Retry: last_exc unset on final attempt — unreachable"

tests/test_retry.py

@@ -520,3 +520,133 @@ async def streamed_files() -> typing.AsyncIterator[bytes]:
         yield b"x"  # pragma: no cover
 
     assert _is_streaming_body(streamed_files()) is True
+
+
+async def test_retry_refuses_streamed_body_request() -> None:
+    """Retry must not replay a request with a streaming body — re-raise with a PEP-678 note."""
+    sleeper = _SleepRecorder()
+    call_count = {"n": 0}
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        call_count["n"] += 1
+        return httpx2.Response(HTTPStatus.SERVICE_UNAVAILABLE, request=request)
+
+    async def streamed_body() -> typing.AsyncIterator[bytes]:
+        yield b"x"
+
+    transport = httpx2.MockTransport(handler)
+    client = AsyncClient(
+        httpx2_client=httpx2.AsyncClient(transport=transport),
+        middleware=[Retry(_sleep=sleeper, base_delay=0.001, max_delay=0.002)],
+    )
+
+    with pytest.raises(ServiceUnavailableError) as info:
+        await client.post("https://example.test/upload", content=streamed_body())
+
+    assert call_count["n"] == 1
+    assert sleeper.calls == []  # no retry attempted
+    notes = getattr(info.value, "__notes__", [])
+    assert any("not retrying" in note and "stream" in note for note in notes)
+
+
+async def test_retry_refuses_streamed_body_does_not_consume_budget() -> None:
+    """When Retry refuses for streaming-body reasons, no budget token is withdrawn."""
+    sleeper = _SleepRecorder()
+    budget = RetryBudget(ttl=10.0, min_retries_per_sec=10.0, percent_can_retry=0.2)
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        return httpx2.Response(HTTPStatus.SERVICE_UNAVAILABLE, request=request)
+
+    async def streamed_body() -> typing.AsyncIterator[bytes]:
+        yield b"x"
+
+    transport = httpx2.MockTransport(handler)
+    client = AsyncClient(
+        httpx2_client=httpx2.AsyncClient(transport=transport),
+        middleware=[Retry(_sleep=sleeper, budget=budget, base_delay=0.001, max_delay=0.002)],
+    )
+
+    with pytest.raises(ServiceUnavailableError):
+        await client.post("https://example.test/upload", content=streamed_body())
+
+    # Budget should be untouched: deposits OK (every attempt deposits), but no withdrawals.
+    # Check via _withdrawn deque emptiness.
+    assert len(budget._withdrawn) == 0  # noqa: SLF001 — implementation-detail access for invariant
+
+
+async def test_retry_refuses_streamed_body_network_error_non_idempotent() -> None:
+    """Streaming POST that hits a NetworkError gets the PEP-678 note."""
+    sleeper = _SleepRecorder()
+
+    def handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        msg = "transient"
+        raise httpx2.ConnectError(msg)
+
+    async def streamed_body() -> typing.AsyncIterator[bytes]:
+        yield b"x"
+
+    transport = httpx2.MockTransport(handler)
+    client = AsyncClient(
+        httpx2_client=httpx2.AsyncClient(transport=transport),
+        middleware=[Retry(_sleep=sleeper, base_delay=0.001, max_delay=0.002)],
+    )
+
+    with pytest.raises(NetworkError) as info:
+        await client.post("https://example.test/upload", content=streamed_body())
+
+    assert sleeper.calls == []  # no retry attempted
+    notes = getattr(info.value, "__notes__", [])
+    assert any("not retrying" in note and "stream" in note for note in notes)
+
+
+async def test_retry_refuses_streamed_body_attempt_timeout_non_idempotent() -> None:
+    """Streaming POST that times out per attempt_timeout gets the PEP-678 note."""
+    sleeper = _SleepRecorder()
+
+    async def slow_handler(request: httpx2.Request) -> httpx2.Response:  # noqa: ARG001
+        await asyncio.sleep(1.0)
+        msg = "should not reach"  # pragma: no cover
+        raise AssertionError(msg)  # pragma: no cover
+
+    async def streamed_body() -> typing.AsyncIterator[bytes]:
+        yield b"x"
+
+    transport = httpx2.MockTransport(slow_handler)
+    client = AsyncClient(
+        httpx2_client=httpx2.AsyncClient(transport=transport),
+        middleware=[Retry(_sleep=sleeper, attempt_timeout=0.05, base_delay=0.001, max_delay=0.002)],
+    )
+
+    with pytest.raises(HttpwareTimeoutError) as info:
+        await client.post("https://example.test/upload", content=streamed_body())
+
+    assert sleeper.calls == []  # no retry attempted
+    notes = getattr(info.value, "__notes__", [])
+    assert any("not retrying" in note and "stream" in note for note in notes)
+
+
+async def test_retry_refuses_streamed_body_idempotent_method() -> None:
+    """Streaming GET that hits a retryable status gets the PEP-678 note instead of retrying."""
+    sleeper = _SleepRecorder()
+    call_count = {"n": 0}
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        call_count["n"] += 1
+        return httpx2.Response(HTTPStatus.SERVICE_UNAVAILABLE, request=request)
+
+    async def streamed_body() -> typing.AsyncIterator[bytes]:
+        yield b"x"
+
+    transport = httpx2.MockTransport(handler)
+    client = AsyncClient(
+        httpx2_client=httpx2.AsyncClient(transport=transport),
+        middleware=[Retry(_sleep=sleeper, base_delay=0.001, max_delay=0.002)],
+    )
+
+    with pytest.raises(ServiceUnavailableError) as info:
+        await client.put("https://example.test/data", content=streamed_body())
+
+    assert call_count["n"] == 1
+    assert sleeper.calls == []  # no retry attempted
+    notes = getattr(info.value, "__notes__", [])
+    assert any("not retrying" in note and "stream" in note for note in notes)