test chatgpt sse output recovery coverage

litellm/responses/sse_output_recovery.py

@@ -79,10 +79,7 @@ def record_output_text_delta_chunk(
     )
     if content_item is None:
         return
-    current_text = content_item.get("text")
-    if not isinstance(current_text, str):
-        current_text = ""
-    content_item["text"] = f"{current_text}{text_delta}"
+    content_item["text"] = f"{content_item['text']}{text_delta}"
 
 
 def record_output_text_chunk(

tests/test_litellm/litellm_core_utils/test_litellm_logging.py

@@ -17,6 +17,7 @@ from litellm.litellm_core_utils.litellm_logging import set_callbacks
 from litellm.types.llms.openai import (
     ResponseAPIUsage,
     ResponseCompletedEvent,
+    ResponsesAPIResponse,
     ResponsesAPIStreamEvents,
 )
 from litellm.types.utils import ModelResponse, TextCompletionResponse
@@ -2204,6 +2205,42 @@ def test_get_assembled_streaming_response_handles_response_event_with_dict_respo
     assert assembled["usage"]["total_tokens"] == 3
 
 
+def test_get_assembled_streaming_response_normalizes_typed_response_usage():
+    import datetime
+
+    logging_obj = _make_logging_obj(stream=True)
+    response = ResponsesAPIResponse(
+        id="resp-1",
+        created_at=1700000000,
+        object="response",
+        status="completed",
+        model="gpt-5.5",
+        output=[],
+        usage=ResponseAPIUsage(
+            input_tokens=4,
+            output_tokens=5,
+            total_tokens=9,
+        ),
+    )
+    result = ResponseCompletedEvent(
+        type=ResponsesAPIStreamEvents.RESPONSE_COMPLETED,
+        response=response,
+    )
+
+    assembled = logging_obj._get_assembled_streaming_response(
+        result=result,
+        start_time=datetime.datetime.now(),
+        end_time=datetime.datetime.now(),
+        is_async=True,
+        streaming_chunks=[],
+    )
+
+    assert assembled is response
+    assert assembled.usage["prompt_tokens"] == 4
+    assert assembled.usage["completion_tokens"] == 5
+    assert assembled.usage["total_tokens"] == 9
+
+
 def test_get_assembled_streaming_response_returns_none_for_non_streaming_text_completion():
     """Non-streaming TextCompletionResponse should also return None."""
     import datetime

tests/test_litellm/responses/test_sse_output_recovery.py

@@ -3,6 +3,7 @@
 from litellm.responses.sse_output_recovery import (
     _MAX_CONTENT_INDEX,
     record_output_text_chunk,
+    record_output_text_delta_chunk,
 )
 
 
@@ -55,3 +56,189 @@ def test_text_chunk_at_max_content_index_is_recorded():
     content = text_only_items[0]["content"]
     assert len(content) == _MAX_CONTENT_INDEX + 1
     assert content[_MAX_CONTENT_INDEX]["text"] == "kept"
+
+
+def test_text_delta_chunks_are_accumulated():
+    output_items: dict = {}
+    text_only_items: dict = {}
+
+    record_output_text_delta_chunk(
+        parsed_chunk={
+            "type": "response.output_text.delta",
+            "output_index": 0,
+            "content_index": 0,
+            "delta": "Hel",
+        },
+        output_items=output_items,
+        text_only_items=text_only_items,
+    )
+    record_output_text_delta_chunk(
+        parsed_chunk={
+            "type": "response.output_text.delta",
+            "output_index": 0,
+            "content_index": 0,
+            "delta": "lo",
+        },
+        output_items=output_items,
+        text_only_items=text_only_items,
+    )
+
+    assert text_only_items[0]["content"][0]["text"] == "Hello"
+
+
+def test_text_delta_with_invalid_delta_is_ignored():
+    output_items: dict = {}
+    text_only_items: dict = {}
+
+    record_output_text_delta_chunk(
+        parsed_chunk={
+            "type": "response.output_text.delta",
+            "output_index": 0,
+            "content_index": 0,
+            "delta": {"not": "text"},
+        },
+        output_items=output_items,
+        text_only_items=text_only_items,
+    )
+
+    assert text_only_items == {}
+
+
+def test_text_delta_does_not_override_output_item_done():
+    output_items = {
+        0: {
+            "type": "message",
+            "role": "assistant",
+            "content": [{"type": "output_text", "text": "final"}],
+        }
+    }
+    text_only_items: dict = {}
+
+    record_output_text_delta_chunk(
+        parsed_chunk={
+            "type": "response.output_text.delta",
+            "output_index": 0,
+            "content_index": 0,
+            "delta": " ignored",
+        },
+        output_items=output_items,
+        text_only_items=text_only_items,
+    )
+
+    assert text_only_items == {}
+    assert output_items[0]["content"][0]["text"] == "final"
+
+
+def test_text_chunk_without_indices_uses_next_available_item_and_content():
+    output_items: dict = {}
+    text_only_items: dict = {}
+
+    record_output_text_chunk(
+        parsed_chunk={
+            "type": "response.output_text.done",
+            "text": "fallback",
+        },
+        output_items=output_items,
+        text_only_items=text_only_items,
+    )
+
+    assert text_only_items[0]["content"][0]["text"] == "fallback"
+
+
+def test_text_chunk_with_invalid_existing_content_is_ignored():
+    output_items: dict = {}
+    text_only_items = {0: {"type": "message", "content": "not-a-list"}}
+
+    record_output_text_chunk(
+        parsed_chunk={
+            "type": "response.output_text.done",
+            "output_index": 0,
+            "content_index": 0,
+            "text": "ignored",
+        },
+        output_items=output_items,
+        text_only_items=text_only_items,
+    )
+
+    assert text_only_items[0]["content"] == "not-a-list"
+
+
+def test_text_chunk_replaces_invalid_existing_content_item():
+    output_items: dict = {}
+    text_only_items = {0: {"type": "message", "content": [None]}}
+
+    record_output_text_chunk(
+        parsed_chunk={
+            "type": "response.output_text.done",
+            "output_index": 0,
+            "content_index": 0,
+            "text": "recovered",
+        },
+        output_items=output_items,
+        text_only_items=text_only_items,
+    )
+
+    assert text_only_items[0]["content"][0]["text"] == "recovered"
+
+
+def test_text_chunk_with_invalid_text_is_ignored():
+    output_items: dict = {}
+    text_only_items: dict = {}
+
+    record_output_text_chunk(
+        parsed_chunk={
+            "type": "response.output_text.done",
+            "output_index": 0,
+            "content_index": 0,
+            "text": None,
+        },
+        output_items=output_items,
+        text_only_items=text_only_items,
+    )
+
+    assert text_only_items == {}
+
+
+def test_text_chunk_does_not_override_output_item_done():
+    output_items = {
+        0: {
+            "type": "message",
+            "role": "assistant",
+            "content": [{"type": "output_text", "text": "authoritative"}],
+        }
+    }
+    text_only_items: dict = {}
+
+    record_output_text_chunk(
+        parsed_chunk={
+            "type": "response.output_text.done",
+            "output_index": 0,
+            "content_index": 0,
+            "text": "ignored",
+        },
+        output_items=output_items,
+        text_only_items=text_only_items,
+    )
+
+    assert text_only_items == {}
+    assert output_items[0]["content"][0]["text"] == "authoritative"
+
+
+def test_text_chunk_preserves_annotations():
+    output_items: dict = {}
+    text_only_items: dict = {}
+    annotations = [{"type": "url_citation", "url": "https://example.com"}]
+
+    record_output_text_chunk(
+        parsed_chunk={
+            "type": "response.output_text.done",
+            "output_index": 0,
+            "content_index": 0,
+            "text": "with annotations",
+            "annotations": annotations,
+        },
+        output_items=output_items,
+        text_only_items=text_only_items,
+    )
+
+    assert text_only_items[0]["content"][0]["annotations"] == annotations