fix(adapters/streaming_iterator.py): Don't send content block after message delta block is sent

Fixes https://github.com/BerriAI/litellm/issues/14315

litellm/constants.py

@@ -15,7 +15,7 @@ DEFAULT_SQS_FLUSH_INTERVAL_SECONDS = int(
     os.getenv("DEFAULT_SQS_FLUSH_INTERVAL_SECONDS", 10)
 )
 DEFAULT_NUM_WORKERS_LITELLM_PROXY = int(
-    os.getenv("DEFAULT_NUM_WORKERS_LITELLM_PROXY", os.cpu_count() or 4)
+    os.getenv("DEFAULT_NUM_WORKERS_LITELLM_PROXY", 1)
 )
 DEFAULT_SQS_BATCH_SIZE = int(os.getenv("DEFAULT_SQS_BATCH_SIZE", 512))
 SQS_SEND_MESSAGE_ACTION = "SendMessage"
@@ -60,7 +60,9 @@ DEFAULT_REASONING_EFFORT_MINIMAL_THINKING_BUDGET_GEMINI_2_5_PRO = int(
     os.getenv("DEFAULT_REASONING_EFFORT_MINIMAL_THINKING_BUDGET_GEMINI_2_5_PRO", 128)
 )
 DEFAULT_REASONING_EFFORT_MINIMAL_THINKING_BUDGET_GEMINI_2_5_FLASH_LITE = int(
-    os.getenv("DEFAULT_REASONING_EFFORT_MINIMAL_THINKING_BUDGET_GEMINI_2_5_FLASH_LITE", 512)
+    os.getenv(
+        "DEFAULT_REASONING_EFFORT_MINIMAL_THINKING_BUDGET_GEMINI_2_5_FLASH_LITE", 512
+    )
 )
 
 # Generic fallback for unknown models
@@ -949,7 +951,9 @@ LITELLM_CLI_SESSION_TOKEN_PREFIX = "litellm-session-token"
 DB_SPEND_UPDATE_JOB_NAME = "db_spend_update_job"
 PROMETHEUS_EMIT_BUDGET_METRICS_JOB_NAME = "prometheus_emit_budget_metrics"
 CLOUDZERO_EXPORT_USAGE_DATA_JOB_NAME = "cloudzero_export_usage_data"
-CLOUDZERO_MAX_FETCHED_DATA_RECORDS = int(os.getenv("CLOUDZERO_MAX_FETCHED_DATA_RECORDS", 50000))
+CLOUDZERO_MAX_FETCHED_DATA_RECORDS = int(
+    os.getenv("CLOUDZERO_MAX_FETCHED_DATA_RECORDS", 50000)
+)
 SPEND_LOG_CLEANUP_JOB_NAME = "spend_log_cleanup"
 SPEND_LOG_RUN_LOOPS = int(os.getenv("SPEND_LOG_RUN_LOOPS", 500))
 SPEND_LOG_CLEANUP_BATCH_SIZE = int(os.getenv("SPEND_LOG_CLEANUP_BATCH_SIZE", 1000))

litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py

@@ -28,10 +28,6 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
         TextBlock,
     )
 
-    def __init__(self, completion_stream: Any, model: str):
-        super().__init__(completion_stream)
-        self.model = model
-
     sent_first_chunk: bool = False
     sent_content_block_start: bool = False
     sent_content_block_finish: bool = False
@@ -39,6 +35,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
     sent_last_message: bool = False
     holding_chunk: Optional[Any] = None
     holding_stop_reason_chunk: Optional[Any] = None
+    queued_usage_chunk: bool = False
     current_content_block_index: int = 0
     current_content_block_start: ContentBlockContentBlockDict = TextBlock(
         type="text",
@@ -47,6 +44,10 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
     pending_new_content_block: bool = False
     chunk_queue: deque = deque()  # Queue for buffering multiple chunks
 
+    def __init__(self, completion_stream: Any, model: str):
+        super().__init__(completion_stream)
+        self.model = model
+
     def __next__(self):
         from .transformation import LiteLLMAnthropicMessagesAdapter
 
@@ -217,77 +218,83 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
 
                     # Queue the merged chunk and reset
                     self.chunk_queue.append(merged_chunk)
+                    self.queued_usage_chunk = True
                     self.holding_stop_reason_chunk = None
                     return self.chunk_queue.popleft()
 
                 # Check if this processed chunk has a stop_reason - hold it for next chunk
 
-                if should_start_new_block and not self.sent_content_block_finish:
-                    # Queue the sequence: content_block_stop -> content_block_start -> current_chunk
+                if not self.queued_usage_chunk:
+                    if should_start_new_block and not self.sent_content_block_finish:
+                        # Queue the sequence: content_block_stop -> content_block_start -> current_chunk
 
-                    # 1. Stop current content block
-                    self.chunk_queue.append(
-                        {
-                            "type": "content_block_stop",
-                            "index": max(self.current_content_block_index - 1, 0),
-                        }
-                    )
+                        # 1. Stop current content block
+                        self.chunk_queue.append(
+                            {
+                                "type": "content_block_stop",
+                                "index": max(self.current_content_block_index - 1, 0),
+                            }
+                        )
 
-                    # 2. Start new content block
-                    self.chunk_queue.append(
-                        {
-                            "type": "content_block_start",
-                            "index": self.current_content_block_index,
-                            "content_block": self.current_content_block_start,
-                        }
-                    )
+                        # 2. Start new content block
+                        self.chunk_queue.append(
+                            {
+                                "type": "content_block_start",
+                                "index": self.current_content_block_index,
+                                "content_block": self.current_content_block_start,
+                            }
+                        )
 
-                    # 3. Queue the current chunk (don't lose it!)
-                    self.chunk_queue.append(processed_chunk)
-
-                    # Reset state for new block
-                    self.sent_content_block_finish = False
-
-                    # Return the first queued item
-                    return self.chunk_queue.popleft()
-
-                if (
-                    processed_chunk["type"] == "message_delta"
-                    and self.sent_content_block_finish is False
-                ):
-                    # Queue both the content_block_stop and the holding chunk
-                    self.chunk_queue.append(
-                        {
-                            "type": "content_block_stop",
-                            "index": self.current_content_block_index,
-                        }
-                    )
-                    self.sent_content_block_finish = True
-                    if processed_chunk.get("delta", {}).get("stop_reason") is not None:
-
-                        self.holding_stop_reason_chunk = processed_chunk
-                    else:
+                        # 3. Queue the current chunk (don't lose it!)
                         self.chunk_queue.append(processed_chunk)
-                    return self.chunk_queue.popleft()
-                elif self.holding_chunk is not None:
-                    # Queue both chunks
-                    self.chunk_queue.append(self.holding_chunk)
-                    self.chunk_queue.append(processed_chunk)
-                    self.holding_chunk = None
-                    return self.chunk_queue.popleft()
-                else:
-                    # Queue the current chunk
-                    self.chunk_queue.append(processed_chunk)
-                    return self.chunk_queue.popleft()
+
+                        # Reset state for new block
+                        self.sent_content_block_finish = False
+
+                        # Return the first queued item
+                        return self.chunk_queue.popleft()
+
+                    if (
+                        processed_chunk["type"] == "message_delta"
+                        and self.sent_content_block_finish is False
+                    ):
+                        # Queue both the content_block_stop and the holding chunk
+                        self.chunk_queue.append(
+                            {
+                                "type": "content_block_stop",
+                                "index": self.current_content_block_index,
+                            }
+                        )
+                        self.sent_content_block_finish = True
+                        if (
+                            processed_chunk.get("delta", {}).get("stop_reason")
+                            is not None
+                        ):
+
+                            self.holding_stop_reason_chunk = processed_chunk
+                        else:
+                            self.chunk_queue.append(processed_chunk)
+                        return self.chunk_queue.popleft()
+                    elif self.holding_chunk is not None:
+                        # Queue both chunks
+                        self.chunk_queue.append(self.holding_chunk)
+                        self.chunk_queue.append(processed_chunk)
+                        self.holding_chunk = None
+                        return self.chunk_queue.popleft()
+                    else:
+                        # Queue the current chunk
+                        self.chunk_queue.append(processed_chunk)
+                        return self.chunk_queue.popleft()
 
             # Handle any remaining held chunks after stream ends
-            if self.holding_stop_reason_chunk is not None:
-                self.chunk_queue.append(self.holding_stop_reason_chunk)
-                self.holding_stop_reason_chunk = None
+            if not self.queued_usage_chunk:
+                if self.holding_stop_reason_chunk is not None:
+                    self.chunk_queue.append(self.holding_stop_reason_chunk)
+                    self.holding_stop_reason_chunk = None
 
-            if self.holding_chunk is not None:
-                self.chunk_queue.append(self.holding_chunk)
-                self.holding_chunk = None
+                if self.holding_chunk is not None:
+                    self.chunk_queue.append(self.holding_chunk)
+                    self.holding_chunk = None
 
             if not self.sent_last_message:
                 self.sent_last_message = True

litellm/proxy/_new_secret_config.yaml

@@ -7,3 +7,6 @@ model_list:
   - model_name: wildcard_models/*
     litellm_params:
       model: openai/*
+  - model_name: xai-grok-3
+    litellm_params:
+      model: xai/grok-3

litellm/proxy/hooks/parallel_request_limiter_v3.py

@@ -6,6 +6,7 @@ This is currently in development and not yet ready for production.
 
 import os
 from datetime import datetime
+from math import floor
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -17,7 +18,7 @@ from typing import (
     Union,
     cast,
 )
-from math import floor
+
 from fastapi import HTTPException
 
 from litellm import DualCache
@@ -95,6 +96,7 @@ end
 return results
 """
 
+
 class RateLimitDescriptorRateLimitObject(TypedDict, total=False):
     requests_per_unit: Optional[int]
     tokens_per_unit: Optional[int]
@@ -480,10 +482,15 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
                     },
                 )
             )
-        
+
         # Team Member rate limits
-        if user_api_key_dict.user_id and (user_api_key_dict.team_member_rpm_limit is not None or user_api_key_dict.team_member_tpm_limit is not None):
-            team_member_value = f"{user_api_key_dict.team_id}:{user_api_key_dict.user_id}"
+        if user_api_key_dict.user_id and (
+            user_api_key_dict.team_member_rpm_limit is not None
+            or user_api_key_dict.team_member_tpm_limit is not None
+        ):
+            team_member_value = (
+                f"{user_api_key_dict.team_id}:{user_api_key_dict.user_id}"
+            )
             descriptors.append(
                 RateLimitDescriptor(
                     key="team_member",
@@ -557,13 +564,13 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
                 # Find which descriptor hit the limit
                 for i, status in enumerate(response["statuses"]):
                     if status["code"] == "OVER_LIMIT":
-                        descriptor = descriptors[floor(i/2)]
+                        descriptor = descriptors[floor(i / 2)]
                         raise HTTPException(
                             status_code=429,
                             detail=f"Rate limit exceeded for {descriptor['key']}: {descriptor['value']}. Remaining: {status['limit_remaining']}",
                             headers={
                                 "retry-after": str(self.window_size),
-                                "rate_limit_type": str(status["rate_limit_type"])
+                                "rate_limit_type": str(status["rate_limit_type"]),
                             },  # Retry after 1 minute
                         )
 
@@ -613,7 +620,9 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
 
         # Check if script is available
         if self.token_increment_script is None:
-            verbose_proxy_logger.debug("TTL preservation script not available, using regular pipeline")
+            verbose_proxy_logger.debug(
+                "TTL preservation script not available, using regular pipeline"
+            )
             await self.internal_usage_cache.dual_cache.async_increment_cache_pipeline(
                 increment_list=pipeline_operations,
                 litellm_parent_otel_span=parent_otel_span,
@@ -628,7 +637,7 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
             for op in pipeline_operations:
                 # Convert None TTL to 0 for Lua script
                 ttl_value = op["ttl"] if op["ttl"] is not None else 0
-                
+
                 verbose_proxy_logger.debug(
                     f"Executing TTL-preserving increment for key={op['key']}, "
                     f"increment={op['increment_value']}, ttl={ttl_value}"
@@ -693,16 +702,15 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
             )
 
             # Get metadata from kwargs
-            user_api_key = kwargs["litellm_params"]["metadata"].get("user_api_key")
-            user_api_key_user_id = kwargs["litellm_params"]["metadata"].get(
-                "user_api_key_user_id"
+            litellm_metadata = kwargs["litellm_params"]["metadata"]
+            if litellm_metadata is None:
+                return
+            user_api_key = litellm_metadata.get("user_api_key")
+            user_api_key_user_id = litellm_metadata.get("user_api_key_user_id")
+            user_api_key_team_id = litellm_metadata.get("user_api_key_team_id")
+            user_api_key_end_user_id = kwargs.get("user") or litellm_metadata.get(
+                "user_api_key_end_user_id"
             )
-            user_api_key_team_id = kwargs["litellm_params"]["metadata"].get(
-                "user_api_key_team_id"
-            )
-            user_api_key_end_user_id = kwargs.get("user") or kwargs["litellm_params"][
-                "metadata"
-            ].get("user_api_key_end_user_id")
             model_group = get_model_group_from_litellm_kwargs(kwargs)
 
             # Get total tokens from response

tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_content_after_stop_reason.py

@@ -0,0 +1,343 @@
+"""
+Test for AnthropicStreamWrapper handling content blocks that exist after message_delta with stop_reason and usage.
+
+This tests the scenario where a streaming response includes:
+1. Initial content blocks
+2. A message_delta chunk with stop_reason and usage
+3. Additional content blocks after the stop_reason
+
+The wrapper should properly handle this by:
+- Holding the stop_reason chunk until usage is available
+- Merging usage into the stop_reason chunk
+- Properly managing content_block_stop/start events for subsequent content
+"""
+
+import os
+import sys
+from typing import List
+
+import pytest
+
+sys.path.insert(0, os.path.abspath("../../../../.."))
+
+from litellm.llms.anthropic.experimental_pass_through.adapters.streaming_iterator import (
+    AnthropicStreamWrapper,
+)
+from litellm.types.utils import Delta, ModelResponse, StreamingChoices, Usage
+
+
+class MockCompletionStreamWithContentAfterStopReason:
+    """Mock stream that simulates content blocks existing after message_delta with stop_reason and usage."""
+
+    def __init__(self):
+        self.responses = [
+            # Initial text content
+            ModelResponse(
+                stream=True,
+                choices=[
+                    StreamingChoices(
+                        delta=Delta(content="Hello"), index=0, finish_reason=None
+                    )
+                ],
+            ),
+            ModelResponse(
+                stream=True,
+                choices=[
+                    StreamingChoices(
+                        delta=Delta(content=" world"), index=0, finish_reason=None
+                    )
+                ],
+            ),
+            # Message delta with stop_reason AND usage (this is how it actually comes from the API)
+            ModelResponse(
+                stream=True,
+                choices=[
+                    StreamingChoices(
+                        delta=Delta(content=""), index=0, finish_reason="stop"
+                    )
+                ],
+                usage=Usage(prompt_tokens=230, completion_tokens=65, total_tokens=295),
+            ),
+            # Additional content after the stop_reason - this simulates the scenario
+            # where there might be additional content blocks after the main response
+            ModelResponse(
+                stream=True,
+                choices=[
+                    StreamingChoices(
+                        delta=Delta(content=" Additional content"),
+                        index=0,
+                        finish_reason=None,
+                    )
+                ],
+            ),
+        ]
+        self.index = 0
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        if self.index >= len(self.responses):
+            raise StopIteration
+        response = self.responses[self.index]
+        self.index += 1
+        return response
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        if self.index >= len(self.responses):
+            raise StopAsyncIteration
+        response = self.responses[self.index]
+        self.index += 1
+        return response
+
+
+def test_anthropic_stream_wrapper_content_after_stop_reason():
+    """Test that AnthropicStreamWrapper properly handles content blocks after message_delta with stop_reason."""
+
+    wrapper = AnthropicStreamWrapper(
+        completion_stream=MockCompletionStreamWithContentAfterStopReason(),
+        model="claude-3",
+    )
+
+    chunks = []
+    chunk_types = []
+
+    # Collect all chunks
+    for chunk in wrapper:
+        chunks.append(chunk)
+        chunk_types.append(chunk.get("type"))
+
+    # Verify the expected sequence of chunk types
+    expected_types = [
+        "message_start",  # Initial message start
+        "content_block_start",  # Start of first content block
+        "content_block_delta",  # "Hello"
+        "content_block_delta",  # " world"
+        "content_block_stop",  # End of first content block due to stop_reason
+        "message_delta",  # Stop reason with merged usage
+        "message_stop",  # Final message stop
+    ]
+
+    print(f"Actual chunk types: {chunk_types}")
+    print(f"Expected chunk types: {expected_types}")
+
+    # Verify we have the expected number of chunks
+    assert len(chunk_types) >= len(
+        expected_types
+    ), f"Expected at least {len(expected_types)} chunks, got {len(chunk_types)}"
+
+    # Verify key chunk types are present
+    assert "message_start" in chunk_types
+    assert "content_block_start" in chunk_types
+    assert "content_block_delta" in chunk_types
+    assert "content_block_stop" in chunk_types
+    assert "message_delta" in chunk_types
+    assert "message_stop" in chunk_types
+
+    # Find the message_delta chunk with stop_reason
+    message_delta_chunk = None
+    for chunk in chunks:
+        if chunk.get("type") == "message_delta":
+            message_delta_chunk = chunk
+            break
+
+    assert message_delta_chunk is not None, "message_delta chunk not found"
+
+    # Verify that the message_delta chunk has both stop_reason and usage
+    delta = message_delta_chunk.get("delta", {})
+    usage = message_delta_chunk.get("usage", {})
+
+    assert (
+        delta.get("stop_reason") == "end_turn"
+    ), f"Expected stop_reason 'end_turn', got {delta.get('stop_reason')}"
+    assert (
+        usage.get("input_tokens") == 230
+    ), f"Expected input_tokens 230, got {usage.get('input_tokens')}"
+    assert (
+        usage.get("output_tokens") == 65
+    ), f"Expected output_tokens 65, got {usage.get('output_tokens')}"
+
+    # Verify content_block_stop comes before message_delta
+    content_block_stop_index = None
+    message_delta_index = None
+
+    for i, chunk_type in enumerate(chunk_types):
+        if chunk_type == "content_block_stop" and content_block_stop_index is None:
+            content_block_stop_index = i
+        elif chunk_type == "message_delta":
+            message_delta_index = i
+
+    assert content_block_stop_index is not None, "content_block_stop not found"
+    assert message_delta_index is not None, "message_delta not found"
+    assert (
+        content_block_stop_index < message_delta_index
+    ), "content_block_stop should come before message_delta"
+
+
+@pytest.mark.asyncio
+async def test_async_anthropic_stream_wrapper_content_after_stop_reason():
+    """Test async version of AnthropicStreamWrapper handling content blocks after message_delta with stop_reason."""
+
+    wrapper = AnthropicStreamWrapper(
+        completion_stream=MockCompletionStreamWithContentAfterStopReason(),
+        model="claude-3",
+    )
+
+    chunks = []
+    chunk_types = []
+
+    # Collect all chunks asynchronously
+    async for chunk in wrapper:
+        chunks.append(chunk)
+        chunk_types.append(chunk.get("type"))
+
+    print(f"Async - Actual chunk types: {chunk_types}")
+
+    # Verify key chunk types are present
+    assert "message_start" in chunk_types
+    assert "content_block_start" in chunk_types
+    assert "content_block_delta" in chunk_types
+    assert "content_block_stop" in chunk_types
+    assert "message_delta" in chunk_types
+    assert "message_stop" in chunk_types
+
+    # Find the message_delta chunk with stop_reason
+    message_delta_chunk = None
+    for chunk in chunks:
+        if chunk.get("type") == "message_delta":
+            message_delta_chunk = chunk
+            break
+
+    assert message_delta_chunk is not None, "message_delta chunk not found"
+
+    # Verify that the message_delta chunk has both stop_reason and usage
+    delta = message_delta_chunk.get("delta", {})
+    usage = message_delta_chunk.get("usage", {})
+
+    assert (
+        delta.get("stop_reason") == "end_turn"
+    ), f"Expected stop_reason 'end_turn', got {delta.get('stop_reason')}"
+    assert (
+        usage.get("input_tokens") == 230
+    ), f"Expected input_tokens 230, got {usage.get('input_tokens')}"
+    assert (
+        usage.get("output_tokens") == 65
+    ), f"Expected output_tokens 65, got {usage.get('output_tokens')}"
+
+
+def test_usage_merging_behavior():
+    """Test that usage information is properly merged with stop_reason chunk."""
+
+    wrapper = AnthropicStreamWrapper(
+        completion_stream=MockCompletionStreamWithContentAfterStopReason(),
+        model="claude-3",
+    )
+
+    # Process chunks and look specifically for the usage merging behavior
+    chunks = []
+    for chunk in wrapper:
+        chunks.append(chunk)
+        # If this is a message_delta with stop_reason, verify it has usage
+        if (
+            chunk.get("type") == "message_delta"
+            and chunk.get("delta", {}).get("stop_reason") is not None
+        ):
+
+            usage = chunk.get("usage", {})
+            assert (
+                usage.get("input_tokens") is not None
+            ), "Usage should be merged with stop_reason chunk"
+            assert (
+                usage.get("output_tokens") is not None
+            ), "Usage should be merged with stop_reason chunk"
+            break
+
+
+def test_sse_wrapper_with_content_after_stop_reason():
+    """Test SSE wrapper formatting for the content after stop_reason scenario."""
+
+    wrapper = AnthropicStreamWrapper(
+        completion_stream=MockCompletionStreamWithContentAfterStopReason(),
+        model="claude-3",
+    )
+
+    # Get SSE formatted chunks
+    sse_chunks = []
+    for chunk in wrapper.anthropic_sse_wrapper():
+        sse_chunks.append(chunk)
+        if len(sse_chunks) >= 10:  # Limit to avoid infinite loops in tests
+            break
+
+    # Verify all chunks are properly formatted as bytes
+    for chunk in sse_chunks:
+        assert isinstance(chunk, bytes), "SSE chunks should be bytes"
+
+        # Decode and verify SSE format
+        chunk_str = chunk.decode("utf-8")
+        lines = chunk_str.split("\n")
+
+        # Should have event and data lines
+        assert any(
+            line.startswith("event: ") for line in lines
+        ), f"Missing event line in: {chunk_str}"
+        assert any(
+            line.startswith("data: ") for line in lines
+        ), f"Missing data line in: {chunk_str}"
+
+
+@pytest.mark.asyncio
+async def test_async_sse_wrapper_with_content_after_stop_reason():
+    """Test async SSE wrapper formatting for the content after stop_reason scenario."""
+
+    wrapper = AnthropicStreamWrapper(
+        completion_stream=MockCompletionStreamWithContentAfterStopReason(),
+        model="claude-3",
+    )
+
+    # Get SSE formatted chunks asynchronously
+    sse_chunks = []
+    async for chunk in wrapper.async_anthropic_sse_wrapper():
+        sse_chunks.append(chunk)
+        if len(sse_chunks) >= 10:  # Limit to avoid infinite loops in tests
+            break
+
+    # Verify all chunks are properly formatted as bytes
+    for chunk in sse_chunks:
+        assert isinstance(chunk, bytes), "Async SSE chunks should be bytes"
+
+        # Decode and verify SSE format
+        chunk_str = chunk.decode("utf-8")
+        lines = chunk_str.split("\n")
+
+        # Should have event and data lines
+        assert any(
+            line.startswith("event: ") for line in lines
+        ), f"Missing event line in: {chunk_str}"
+        assert any(
+            line.startswith("data: ") for line in lines
+        ), f"Missing data line in: {chunk_str}"
+
+
+if __name__ == "__main__":
+    # Run a quick test
+    test_anthropic_stream_wrapper_content_after_stop_reason()
+    print("✅ Sync test passed")
+
+    import asyncio
+
+    asyncio.run(test_async_anthropic_stream_wrapper_content_after_stop_reason())
+    print("✅ Async test passed")
+
+    test_usage_merging_behavior()
+    print("✅ Usage merging test passed")
+
+    test_sse_wrapper_with_content_after_stop_reason()
+    print("✅ SSE wrapper test passed")
+
+    asyncio.run(test_async_sse_wrapper_with_content_after_stop_reason())
+    print("✅ Async SSE wrapper test passed")
+
+    print("🎉 All tests passed!")