feat(client): mark requests with async-iterable bodies via extensions

Adds a _is_streaming_body helper and a marker step in _request_with_body:
when content / data / files is an async-iterable, set
request.extensions['httpware.streaming_body'] = True before sending.

Sets up Task 3: Retry will read the marker and refuse to retry streamed-body
requests (they can't replay across attempts). Today the marker has no
consumer; it's harmless metadata.

Author: lesnik512

src/httpware/client.py

@@ -71,6 +71,15 @@ def _raise_on_status_error(response: httpx2.Response) -> None:
         raise exc_class(response)
 
 
+def _is_streaming_body(value: typing.Any) -> bool:
+    """Return True if value is an async-iterable that cannot be safely replayed for retry."""
+    if value is None:
+        return False
+    if isinstance(value, (bytes, bytearray, memoryview, str, dict)):
+        return False
+    return hasattr(value, "__aiter__")
+
+
 class AsyncClient:
     """Async HTTP client: thin wrapper around httpx2 with typed decoding and middleware."""
 
@@ -164,7 +173,7 @@ def build_request(self, method: str, url: str, **kwargs: typing.Any) -> httpx2.R
         """Delegate request construction to the wrapped httpx2.AsyncClient."""
         return self._httpx2_client.build_request(method, url, **kwargs)
 
-    async def _request_with_body(  # noqa: PLR0913 — mirrors httpx2 per-method signatures
+    async def _request_with_body(  # noqa: PLR0913, C901 — mirrors httpx2 per-method signatures; kwargs-forwarding complexity is structural
         self,
         method: str,
         url: str,
@@ -200,6 +209,8 @@ async def _request_with_body(  # noqa: PLR0913 — mirrors httpx2 per-method sig
         if files is not None:
             kwargs["files"] = files
         request = self._httpx2_client.build_request(method, url, **kwargs)
+        if _is_streaming_body(content) or _is_streaming_body(data) or _is_streaming_body(files):
+            request.extensions["httpware.streaming_body"] = True
         return await self.send(request, response_model=response_model)
 
     @typing.overload

tests/test_retry.py

@@ -7,13 +7,15 @@
 import asyncio
 import datetime
 import email.utils
+import typing
 from collections.abc import Callable
 from http import HTTPStatus
 
 import httpx2
 import pytest
 
 from httpware import AsyncClient, NotFoundError, ServiceUnavailableError, TransportError
+from httpware.client import _is_streaming_body
 from httpware.errors import NetworkError, RetryBudgetExhaustedError
 from httpware.errors import TimeoutError as HttpwareTimeoutError
 from httpware.middleware.resilience.budget import RetryBudget
@@ -441,3 +443,80 @@ async def test_explicit_budget_shared_across_retry_instances() -> None:
     for _ in range(10):
         assert shared.try_withdraw() is True
     assert shared.try_withdraw() is False
+
+
+async def test_client_post_with_async_iterable_content_marks_extensions() -> None:
+    """Posting with an async-iterable body sets the httpware.streaming_body marker on request.extensions."""
+    seen_extensions: list[dict[str, object]] = []
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        seen_extensions.append(dict(request.extensions))
+        return httpx2.Response(HTTPStatus.OK, request=request)
+
+    async def streamed_body() -> typing.AsyncIterator[bytes]:
+        yield b"chunk1"
+        yield b"chunk2"
+
+    transport = httpx2.MockTransport(handler)
+    client = AsyncClient(httpx2_client=httpx2.AsyncClient(transport=transport))
+    await client.post("https://example.test/upload", content=streamed_body())
+
+    assert len(seen_extensions) == 1
+    assert seen_extensions[0].get("httpware.streaming_body") is True
+
+
+async def test_client_post_with_bytes_content_does_not_mark_extensions() -> None:
+    seen_extensions: list[dict[str, object]] = []
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        seen_extensions.append(dict(request.extensions))
+        return httpx2.Response(HTTPStatus.OK, request=request)
+
+    transport = httpx2.MockTransport(handler)
+    client = AsyncClient(httpx2_client=httpx2.AsyncClient(transport=transport))
+    await client.post("https://example.test/upload", content=b"hi")
+
+    assert len(seen_extensions) == 1
+    assert "httpware.streaming_body" not in seen_extensions[0]
+
+
+async def test_client_post_with_dict_data_does_not_mark_extensions() -> None:
+    seen_extensions: list[dict[str, object]] = []
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        seen_extensions.append(dict(request.extensions))
+        return httpx2.Response(HTTPStatus.OK, request=request)
+
+    transport = httpx2.MockTransport(handler)
+    client = AsyncClient(httpx2_client=httpx2.AsyncClient(transport=transport))
+    await client.post("https://example.test/upload", data={"k": "v"})
+
+    assert len(seen_extensions) == 1
+    assert "httpware.streaming_body" not in seen_extensions[0]
+
+
+async def test_client_post_with_async_iterable_data_marks_extensions() -> None:
+    seen_extensions: list[dict[str, object]] = []
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        seen_extensions.append(dict(request.extensions))
+        return httpx2.Response(HTTPStatus.OK, request=request)
+
+    async def streamed_data() -> typing.AsyncIterator[bytes]:
+        yield b"x"
+
+    transport = httpx2.MockTransport(handler)
+    client = AsyncClient(httpx2_client=httpx2.AsyncClient(transport=transport))
+    await client.post("https://example.test/upload", data=streamed_data())
+
+    assert len(seen_extensions) == 1
+    assert seen_extensions[0].get("httpware.streaming_body") is True
+
+
+def test_is_streaming_body_true_for_async_iterable_files() -> None:
+    """_is_streaming_body returns True for an async-iterable, covering the files= path."""
+
+    async def streamed_files() -> typing.AsyncIterator[bytes]:
+        yield b"x"  # pragma: no cover
+
+    assert _is_streaming_body(streamed_files()) is True