fix: resolve streaming duplication and markdown formatting issues

- Fix text duplication bug with proper delta calculation
- Preserve markdown formatting with chunk-based delivery (50 chars)
- Handle GeneratorExit errors from async context managers
- Update Chat service streaming to preserve formatting
- Ensure proper word-by-word streaming without duplicates

src/chat/service.py

@@ -214,9 +214,12 @@ async def create_chat_completion_stream(
             in_reasoning = False
 
         elif item.type == "message":
-            # Stream message content word by word
+            # Stream message content in chunks (preserves newlines, markdown, etc.)
             text = item.data["content"][0]["text"]
-            for word in text.split():
+            chunk_size = 50  # characters per chunk
+
+            for i in range(0, len(text), chunk_size):
+                chunk = text[i:i+chunk_size]
                 yield ChatCompletionChunk(
                     id=completion_id,
                     object=constants.CHAT_COMPLETION_CHUNK_OBJECT,
@@ -225,12 +228,12 @@ async def create_chat_completion_stream(
                     choices=[
                         ChatCompletionChunkChoice(
                             index=0,
-                            delta=ChatCompletionChunkDelta(content=f"{word} "),
+                            delta=ChatCompletionChunkDelta(content=chunk),
                             finish_reason=None,
                         )
                     ],
                 )
-                await asyncio.sleep(0.05)  # Simulate typing
+                await asyncio.sleep(0.02)  # Faster since chunks are larger
 
     # Final chunk with finish_reason
     yield ChatCompletionChunk(

src/responses/streaming.py

@@ -140,6 +140,7 @@ class StreamingCoordinator:
         import asyncio
 
         output_items = []
+        last_message_text = ""  # Track last streamed message text to compute deltas
 
         try:
             # Strip pipeline prefix if present (e.g., "pipeline.model" -> "model")
@@ -189,45 +190,51 @@ class StreamingCoordinator:
                     yield FunctionCallDone()
 
                 elif item.type == "message":
-                    # Stream output text with stop sequence and max tokens enforcement
-                    text = item.data["content"][0]["text"]
-                    words = text.split()
+                    # Get current accumulated text from agent
+                    current_text = item.data["content"][0]["text"]
 
-                    # Track accumulated text and tokens for enforcement
-                    accumulated_text = ""
-                    output_tokens = 0
+                    # Only stream the NEW text (delta) since last update
+                    if current_text.startswith(last_message_text):
+                        # Extract only the new portion
+                        delta_text = current_text[len(last_message_text):]
 
-                    for word in words:
-                        # Add word to accumulated text
-                        word_with_space = f"{word} "
-                        accumulated_text += word_with_space
+                        if delta_text:
+                            # Stream the delta text in chunks while preserving formatting
+                            # (newlines, markdown, code blocks, etc.)
+                            chunk_size = 50  # characters per chunk
 
-                        # Check stop sequences
-                        stop_found, text_before_stop = self._check_stop_sequence(
-                            accumulated_text,
-                            request.stop
-                        )
+                            for i in range(0, len(delta_text), chunk_size):
+                                chunk = delta_text[i:i+chunk_size]
 
-                        if stop_found:
-                            # Emit final text before stop sequence
-                            remaining_text = text_before_stop[len(accumulated_text) - len(word_with_space):]
-                            if remaining_text:
-                                yield OutputTextDelta(delta=remaining_text)
-                            yield OutputTextDone()
-                            break
+                                # Check stop sequences on full accumulated text
+                                stop_found, text_before_stop = self._check_stop_sequence(
+                                    current_text,
+                                    request.stop
+                                )
 
-                        # Check max tokens
-                        output_tokens = self._count_tokens_approx(accumulated_text)
-                        if self._check_max_tokens(output_tokens, request.max_output_tokens):
-                            # Max tokens reached - stop streaming
-                            yield OutputTextDone()
-                            break
+                                if stop_found:
+                                    # Only emit remaining delta before stop
+                                    remaining = text_before_stop[len(last_message_text):]
+                                    if remaining:
+                                        yield OutputTextDelta(delta=remaining)
+                                    yield OutputTextDone()
+                                    break
 
-                        # Normal streaming
-                        yield OutputTextDelta(delta=word_with_space)
-                        await asyncio.sleep(0.05)  # Simulate typing
-                    else:
-                        # Completed normally without stop/limit
+                                # Check max tokens on full text
+                                output_tokens = self._count_tokens_approx(current_text)
+                                if self._check_max_tokens(output_tokens, request.max_output_tokens):
+                                    yield OutputTextDone()
+                                    break
+
+                                # Normal streaming of delta chunk (preserves all formatting)
+                                yield OutputTextDelta(delta=chunk)
+                                await asyncio.sleep(0.02)  # Shorter delay since chunks are larger
+
+                        # Update tracking variable
+                        last_message_text = current_text
+
+                    # If this is the final message (status=completed), ensure we send done
+                    if item.data.get("status") == "completed":
                         yield OutputTextDone()
 
             # Final response.done event with complete response