Merge branch 'fix/streaming-response'

- Thread-safe BaseStreamResponse with sync/async mode
- Fix _stream_generate deadlock (async→sync)
- Refactor OpenAI provider response processing

Author: vilsonrodrigues

src/msgflux/_private/response.py

@@ -1,5 +1,7 @@
 import asyncio
-from typing import Any, AsyncGenerator, Union
+import threading
+from collections import deque
+from typing import Any, AsyncGenerator, Literal, Union
 
 
 class CoreResponse:
@@ -29,24 +31,52 @@ def consume(self) -> Any:
 
 
 class BaseStreamResponse(CoreResponse):
-    def __init__(self):
-        self.first_chunk_event = asyncio.Event()
+    def __init__(self, mode: Literal["sync", "async"] = "sync"):
+        if mode not in {"sync", "async"}:
+            raise ValueError("`mode` must be `sync` or `async`")
+        self.mode = mode
+        if mode == "async":
+            self.first_chunk_event = asyncio.Event()
+        else:
+            self.first_chunk_event = threading.Event()
         self.data = None
-        self.queue = asyncio.Queue()
+        self._queue = None
+        self._queue_loop = None
+        self._pending_chunks = deque()
+        self._queue_lock = threading.Lock()
         self.metadata = None
         self.response_type = None
 
     def add(self, data: Any):
-        """Add data to the stream queue (async)."""
-        self.queue.put_nowait(data)
+        """Add data to the stream queue in a thread-safe way."""
+        with self._queue_lock:
+            queue = self._queue
+            loop = self._queue_loop
+            if queue is None or loop is None or loop.is_closed():
+                self._pending_chunks.append(data)
+                return
+
+        loop.call_soon_threadsafe(queue.put_nowait, data)
+
+    def _bind_consumer_queue(self) -> asyncio.Queue:
+        loop = asyncio.get_running_loop()
+        with self._queue_lock:
+            if self._queue is None:
+                self._queue = asyncio.Queue()
+                self._queue_loop = loop
+                while self._pending_chunks:
+                    self._queue.put_nowait(self._pending_chunks.popleft())
+            elif self._queue_loop is not loop:
+                raise RuntimeError(
+                    "BaseStreamResponse.consume() must run on the same event loop."
+                )
+            return self._queue
 
     async def consume(self) -> AsyncGenerator[Union[bytes, str], None]:
         """Async generator that yields chunks from the queue until None is received."""
+        queue = self._bind_consumer_queue()
         while True:
-            try:
-                chunk = await asyncio.wait_for(self.queue.get(), timeout=1.0)
-                if chunk is None:
-                    break
-                yield chunk
-            except asyncio.TimeoutError:
-                continue
+            chunk = await queue.get()
+            if chunk is None:
+                break
+            yield chunk