From 654d07611eda2c6158a0a556c0e77f2ca23f9b19 Mon Sep 17 00:00:00 2001 From: Bennytimz Date: Fri, 3 Jul 2026 03:03:10 +0100 Subject: [PATCH] fix(client): handle multipart and streaming bodies on paid 402 retry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Complements PR #156 (bytes/content= body replay) with two additions: 1. Multipart / files= bodies: aread() is called before the first dispatch, buffering the MultipartStream into request._content. The paid retry uses content=request.content so the full multipart payload is replayed. Tests: test_paid_retry_replays_multipart_body, test_paid_retry_replays_put_body, test_paid_retry_replays_patch_body. 2. Async generator bodies: an AsyncByteStream that is not also a SyncByteStream (i.e. AsyncIteratorByteStream from content=async_gen()) cannot be safely buffered for replay — the generator may be infinite, already partially consumed, or tied to a one-shot I/O source. PaymentTransport now raises PaymentError immediately with a clear message directing callers to use a buffered body instead. Test: test_paid_retry_raises_for_streaming_body. Also includes the core fix from #156: await request.aread() before dispatch and content=request.content (not stream=request.stream) on retry, so that all finite body types are correctly replayed. --- .changelog/multipart-retry-body.md | 5 + src/mpp/client/transport.py | 22 +++- tests/test_client.py | 160 +++++++++++++++++++++++++++++ 3 files changed, 186 insertions(+), 1 deletion(-) create mode 100644 .changelog/multipart-retry-body.md diff --git a/.changelog/multipart-retry-body.md b/.changelog/multipart-retry-body.md new file mode 100644 index 0000000..d06904f --- /dev/null +++ b/.changelog/multipart-retry-body.md @@ -0,0 +1,5 @@ +--- +pympp: patch +--- + +Handle multipart (`files=`) and streaming bodies on paid 402 retry. Multipart bodies are buffered and replayed identically; async generator bodies raise `PaymentError` before any I/O. diff --git a/src/mpp/client/transport.py b/src/mpp/client/transport.py index 3fa83db..5712f01 100644 --- a/src/mpp/client/transport.py +++ b/src/mpp/client/transport.py @@ -17,6 +17,7 @@ from mpp import Challenge, Credential from mpp._parsing import ParseError +from mpp.errors import PaymentError from mpp.events import ( CHALLENGE_RECEIVED, CREDENTIAL_CREATED, @@ -117,6 +118,25 @@ def on_payment_failed(self, handler: EventHandler) -> Unsubscribe: async def handle_async_request(self, request: httpx.Request) -> httpx.Response: """Handle request, automatically retrying on 402 with credentials.""" + # Async-generator bodies (content=async_gen()) produce an AsyncByteStream + # that is not also a SyncByteStream. They cannot be safely buffered for + # replay: the generator may be infinite, already partially consumed, or + # tied to a one-shot I/O source. Reject early so callers get a clear + # message instead of a silent empty body on the paid retry. + if isinstance(request.stream, httpx.AsyncByteStream) and not isinstance( + request.stream, httpx.SyncByteStream + ): + raise PaymentError( + "Streaming request bodies (async generators) are not supported " + "through the payment retry flow. Use a buffered body " + "(bytes, str, files=, or data=) instead." + ) + + # Buffer the request body before the first dispatch so it can be replayed + # on a paid 402 retry. Handles bytes bodies and multipart/files= bodies. + # After aread() the stream is replaced with a replayable ByteStream. + await request.aread() + response = await self._inner.handle_async_request(request) if response.status_code != 402: @@ -240,7 +260,7 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response: method=request.method, url=request.url, headers=headers, - stream=request.stream, + content=request.content, extensions=request.extensions, ) diff --git a/tests/test_client.py b/tests/test_client.py index 2164399..c13de90 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -89,6 +89,166 @@ async def test_handles_402_with_matching_method(self) -> None: method.create_credential.assert_called_once() + @pytest.mark.asyncio + async def test_paid_retry_replays_request_body(self) -> None: + """The paid retry must carry the original bytes request body. + + Regression: the retry previously reused the already-consumed request + stream, so a POST body was dropped on the (paying) retry. + """ + challenge = Challenge( + id="test-id", + method="tempo", + intent="charge", + request={"amount": "1000"}, + ) + www_auth = challenge.to_www_authenticate("example.com") + + inner = MockTransport( + [ + httpx.Response(402, headers={"www-authenticate": www_auth}), + httpx.Response(200, content=b'{"data": "ok"}'), + ] + ) + transport = PaymentTransport(methods=[MockMethod()], inner=inner) + + body = b'{"prompt": "expensive question"}' + request = httpx.Request("POST", "https://example.com", content=body) + await transport.handle_async_request(request) + + assert len(inner.requests) == 2 + retry_request = inner.requests[1] + assert retry_request.content == body + assert retry_request.headers["content-length"] == str(len(body)) + + @pytest.mark.asyncio + async def test_paid_retry_replays_multipart_body(self) -> None: + """The paid retry must carry the original multipart (files=) body.""" + challenge = Challenge( + id="test-id", + method="tempo", + intent="charge", + request={"amount": "1000"}, + ) + www_auth = challenge.to_www_authenticate("example.com") + + inner = MockTransport( + [ + httpx.Response(402, headers={"www-authenticate": www_auth}), + httpx.Response(200, content=b'{"data": "ok"}'), + ] + ) + transport = PaymentTransport(methods=[MockMethod()], inner=inner) + + file_content = b"hello from file" + request = httpx.Request( + "POST", + "https://example.com", + files={"upload": ("report.txt", file_content, "text/plain")}, + ) + await transport.handle_async_request(request) + + assert len(inner.requests) == 2 + initial_body = inner.requests[0].content + retry_body = inner.requests[1].content + assert retry_body == initial_body + assert b"hello from file" in retry_body + assert b"report.txt" in retry_body + + @pytest.mark.asyncio + async def test_paid_retry_replays_put_body(self) -> None: + """The paid retry must carry the original body for PUT requests.""" + challenge = Challenge( + id="test-id", + method="tempo", + intent="charge", + request={"amount": "1000"}, + ) + www_auth = challenge.to_www_authenticate("example.com") + + inner = MockTransport( + [ + httpx.Response(402, headers={"www-authenticate": www_auth}), + httpx.Response(200, content=b'{"updated": true}'), + ] + ) + transport = PaymentTransport(methods=[MockMethod()], inner=inner) + + body = b'{"name": "updated"}' + request = httpx.Request("PUT", "https://example.com/item/1", content=body) + await transport.handle_async_request(request) + + assert len(inner.requests) == 2 + retry_request = inner.requests[1] + assert retry_request.method == "PUT" + assert retry_request.content == body + assert retry_request.headers["content-length"] == str(len(body)) + + @pytest.mark.asyncio + async def test_paid_retry_replays_patch_body(self) -> None: + """The paid retry must carry the original body for PATCH requests.""" + challenge = Challenge( + id="test-id", + method="tempo", + intent="charge", + request={"amount": "1000"}, + ) + www_auth = challenge.to_www_authenticate("example.com") + + inner = MockTransport( + [ + httpx.Response(402, headers={"www-authenticate": www_auth}), + httpx.Response(200, content=b'{"patched": true}'), + ] + ) + transport = PaymentTransport(methods=[MockMethod()], inner=inner) + + body = b'{"status": "active"}' + request = httpx.Request("PATCH", "https://example.com/item/1", content=body) + await transport.handle_async_request(request) + + assert len(inner.requests) == 2 + retry_request = inner.requests[1] + assert retry_request.method == "PATCH" + assert retry_request.content == body + assert retry_request.headers["content-length"] == str(len(body)) + + @pytest.mark.asyncio + async def test_paid_retry_raises_for_streaming_body(self) -> None: + """Should raise PaymentError upfront for async generator (streaming) bodies. + + Streaming bodies cannot be reliably buffered and replayed: the generator + may be infinite, already partially consumed, or tied to a one-shot source. + A descriptive error is better than a silent empty body on the paid retry. + """ + from mpp.errors import PaymentError + + challenge = Challenge( + id="test-id", + method="tempo", + intent="charge", + request={"amount": "1000"}, + ) + www_auth = challenge.to_www_authenticate("example.com") + + inner = MockTransport( + [ + httpx.Response(402, headers={"www-authenticate": www_auth}), + httpx.Response(200, content=b'{"data": "ok"}'), + ] + ) + transport = PaymentTransport(methods=[MockMethod()], inner=inner) + + async def async_body_gen(): + yield b"chunk1" + yield b"chunk2" + + request = httpx.Request("POST", "https://example.com", content=async_body_gen()) + with pytest.raises(PaymentError, match="Streaming request bodies"): + await transport.handle_async_request(request) + + assert len(inner.requests) == 0 + @pytest.mark.asyncio async def test_emits_client_payment_events(self) -> None: """Should emit mppx-compatible client payment lifecycle events."""