Harden the streaming ASGI bridge against trailing responses

Once an application sends the final http.response.body chunk, the bridge
now drops any further http.response.start/body messages, matching what a
real ASGI server's client observes. Starlette's request_response sends a
trailing response when an endpoint's sub-application has already completed
a rejection response, so the SSE security rejection tests no longer depend
on scheduling for the client to read the first status. Pinned by two new
bridge contract tests (registered as harness self-tests).

Also widen the trio-leg unraisable-warning filter to the whole
httpx/httpx-sse generator chain: abandoning EventSource.aiter_sse abandons
the nested aiter_lines -> aiter_text -> aiter_bytes -> aiter_raw
generators, and which link the finalizer reports depends on GC timing and
Python version.

Author: maxisbey

tests/interaction/conftest.py

@@ -27,14 +27,17 @@ def pytest_configure(config: pytest.Config) -> None:
     # only test on the trio backend. v1's streamable-HTTP client abandons its httpx/httpx-sse
     # response generators when the session task group is cancelled at teardown; asyncio finalizes
     # abandoned async generators silently at loop shutdown, but trio's finalizer warns about each
-    # one (`Async generator ... was garbage collected before it had been exhausted`). The fixes
-    # live in `src/` on `main` and are out of scope for this tests-only backport. The filters are
-    # scoped to the two known httpx generator signatures so an unrelated leak still fails the suite.
+    # one (`Async generator ... was garbage collected before it had been exhausted`). Abandoning
+    # `EventSource.aiter_sse` abandons the whole generator chain nested under it (`aiter_lines` ->
+    # `aiter_text` -> `aiter_bytes` -> `aiter_raw`), and which links the finalizer reports depends
+    # on GC timing and Python version. The fixes live in `src/` on `main` and are out of scope for
+    # this tests-only backport. The filters are scoped to the httpx/httpx-sse generator signatures
+    # (every generator in that chain lives on `Response` or `EventSource`) so an unrelated leak
+    # still fails the suite.
     config.addinivalue_line("filterwarnings", "ignore:Async generator 'httpx:ResourceWarning")
     config.addinivalue_line(
         "filterwarnings",
-        "ignore:.*async_generator object (Response.aiter_text|EventSource.aiter_sse)"
-        ":pytest.PytestUnraisableExceptionWarning",
+        "ignore:.*async_generator object (Response|EventSource).aiter_:pytest.PytestUnraisableExceptionWarning",
     )
 
 

tests/interaction/test_coverage.py

@@ -27,6 +27,8 @@
 _HARNESS_SELF_TESTS = {
     "tests.interaction.lowlevel.test_wire.test_recording_read_stream_ends_iteration_when_the_sender_closes",
     "tests.interaction.transports.test_bridge.test_response_chunks_arrive_as_the_application_sends_them",
+    "tests.interaction.transports.test_bridge.test_a_second_response_after_the_first_completes_is_invisible_to_the_client",
+    "tests.interaction.transports.test_bridge.test_body_chunks_after_the_final_chunk_are_ignored",
     "tests.interaction.transports.test_bridge.test_closing_the_response_delivers_a_disconnect_to_the_application",
     "tests.interaction.transports.test_bridge.test_an_application_failure_before_the_response_starts_fails_the_request",
     "tests.interaction.transports.test_bridge.test_disabling_cancel_on_close_lets_the_application_finish_after_disconnect",

tests/interaction/transports/_bridge.py

@@ -12,6 +12,8 @@
 
 - The request body is buffered before the application is invoked (MCP requests are small JSON
   documents); the response streams chunk by chunk.
+- The response ends at the first `http.response.body` whose `more_body` is falsy; anything the
+  application sends after that is ignored, exactly as a real server's client never observes it.
 - Closing the response — or the whole client — delivers `http.disconnect` to the application,
   exactly as a real server sees when its peer goes away.
 - An exception the application raises before sending `http.response.start` fails the originating
@@ -116,6 +118,7 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
         request_delivered = False
         client_disconnected = anyio.Event()
         response_started = anyio.Event()
+        response_complete = False
         response_status = 0
         response_headers: list[tuple[bytes, bytes]] = []
         application_error: Exception | None = None
@@ -130,7 +133,14 @@ async def receive_request() -> Message:
             return {"type": "http.disconnect"}
 
         async def send_response(message: Message) -> None:
-            nonlocal response_status, response_headers
+            nonlocal response_complete, response_status, response_headers
+            if response_complete:
+                # The response ended with the final body chunk below; a real server's client never
+                # observes anything sent after that, so drop it. Starlette's `request_response`
+                # makes this path real: an endpoint whose sub-application already sent a complete
+                # rejection response (the legacy SSE transport's request validation) still returns
+                # a `Response`, which sends a trailing second start/body pair.
+                return
             if message["type"] == "http.response.start":
                 response_status = message["status"]
                 response_headers = list(message.get("headers", []))
@@ -141,6 +151,7 @@ async def send_response(message: Message) -> None:
             if body:
                 await chunk_writer.send(body)
             if not message.get("more_body", False):
+                response_complete = True
                 await chunk_writer.aclose()
 
         async def run_application() -> None:

tests/interaction/transports/test_bridge.py

@@ -40,6 +40,53 @@ async def chunked_app(scope: Scope, receive: Receive, send: Send) -> None:
     assert chunks == [b"first", b"second"]
 
 
+async def test_a_second_response_after_the_first_completes_is_invisible_to_the_client() -> None:
+    """Only the first complete response reaches the client; a trailing start/body pair is dropped.
+
+    Starlette's `request_response` produces exactly this sequence when an endpoint's
+    sub-application has already sent a complete rejection response (the legacy SSE transport's
+    request validation): the endpoint still returns a `Response`, which sends a second response.
+    """
+
+    async def double_responding_app(scope: Scope, receive: Receive, send: Send) -> None:
+        assert scope["type"] == "http"
+        assert (await receive())["type"] == "http.request"
+        await send({"type": "http.response.start", "status": 421, "headers": [(b"content-type", b"text/plain")]})
+        await send({"type": "http.response.body", "body": b"rejected", "more_body": False})
+        await send({"type": "http.response.start", "status": 200, "headers": [(b"x-late", b"yes")]})
+        await send({"type": "http.response.body", "body": b"too late", "more_body": False})
+
+    transport = StreamingASGITransport(double_responding_app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://bridge") as http:
+        response = await http.get("/double")
+
+    assert response.status_code == 421
+    assert response.text == "rejected"
+    assert "x-late" not in response.headers
+
+
+async def test_body_chunks_after_the_final_chunk_are_ignored() -> None:
+    """Extra body chunks after `more_body: False` neither reach the client nor fail the application."""
+    application_finished = anyio.Event()
+
+    async def overflowing_app(scope: Scope, receive: Receive, send: Send) -> None:
+        assert scope["type"] == "http"
+        assert (await receive())["type"] == "http.request"
+        await send({"type": "http.response.start", "status": 200, "headers": []})
+        await send({"type": "http.response.body", "body": b"complete", "more_body": False})
+        await send({"type": "http.response.body", "body": b"overflow", "more_body": True})
+        application_finished.set()
+
+    transport = StreamingASGITransport(overflowing_app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://bridge") as http:
+        response = await http.get("/overflow")
+        with anyio.fail_after(5):
+            await application_finished.wait()
+
+    assert response.status_code == 200
+    assert response.text == "complete"
+
+
 async def test_closing_the_response_delivers_a_disconnect_to_the_application() -> None:
     """A client that closes the response early is seen by the application as an http.disconnect."""
     seen_after_request: list[Message] = []