test(relay): the OpenAI-stream e2e no longer self-deadlocks when the stream is buffered

Under load the fake provider stream is exhausted before the consumer pulls
its first chunk, and Relay then runs its finalizer inline on the stream loop:
on the consumer thread, inside __next__. The test made that finalizer wait
for the consumer to release it, which the same thread can never do (30 s),
and its assertion sat outside the finally that marks the finalizer done, so
the consumer then waited another 30 s. A monotonic timeline from a loaded
repro showed exactly that order. The finalizer now skips the wait when it
runs on the stream loop (the forced ordering already holds there), and the
wait sits inside the try. Loaded repro: 4/40 first-attempt failures before,
0/80 after.

tests/e2e/test_relay_native_openai_stream.py

@@ -19,6 +19,15 @@ def _sse(*chunk_bodies: bytes) -> bytes:
     return b"".join(_CHUNK_PREFIX + body + b"}\n\n" for body in chunk_bodies) + b"data: [DONE]\n\n"
 
 
+def _running_on(loop) -> bool:
+    import asyncio
+
+    try:
+        return asyncio.get_running_loop() is loop
+    except RuntimeError:
+        return False
+
+
 def _stream_through_relay(tmp_path, monkeypatch, response_body: bytes, *, finalize_before):
     """Stream ``response_body`` through Relay; Relay's finalizer is forced to complete before
     the consumer thread processes the first chunk matching ``finalize_before(chunk)``.
@@ -60,8 +69,13 @@ def _stream_through_relay(tmp_path, monkeypatch, response_body: bytes, *, finali
 
     def run_synchronized_relay_finalizer(managed_stream, attempt):
         relay_finalizer_started.set()
-        assert allow_relay_finalizer.wait(30), "consumer did not release Relay's finalizer"
         try:
+            # When the provider stream is exhausted before the consumer pulls its first chunk, Relay
+            # runs the finalizer inline on the stream's own loop, i.e. on the consumer thread inside
+            # ``__next__``: the consumer cannot release it from there, and the forced ordering
+            # (finalizer before the chosen chunk) already holds. Waiting would self-deadlock.
+            if not _running_on(managed_stream._loop):
+                assert allow_relay_finalizer.wait(30), "consumer did not release Relay's finalizer"
             return run_relay_finalizer(managed_stream, attempt)
         finally:
             relay_finalizer_finished.set()