fix(ai): capture usage and output of incomplete Responses streams

A Responses API stream cut short, for example by max_output_tokens, ends on
response.incomplete instead of response.completed. The stream state only read
usage and output from response.completed, so these generations were captured
with no token counts and empty output even though the stop reason was set.

Read them from every terminal event (completed, incomplete, failed), as the
stop reason already does and as posthog-js does.

Author: breken-ai

.sampo/changesets/responses-incomplete-stream-usage.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: patch
+---
+
+Capture token usage and output for OpenAI Responses streams that end incomplete, such as when `max_output_tokens` is reached.

posthog/ai/openai/openai_converter.py

@@ -520,6 +520,15 @@ def extract_openai_usage_from_response(response: Any) -> TokenUsage:
     return result
 
 
+# Stream events that carry the final Responses API response, with its usage and
+# output. Exactly one of them ends a stream.
+_RESPONSES_TERMINAL_EVENT_TYPES = (
+    "response.completed",
+    "response.incomplete",
+    "response.failed",
+)
+
+
 def extract_openai_usage_from_chunk(
     chunk: Any, provider_type: str = "chat"
 ) -> TokenUsage:
@@ -577,8 +586,10 @@ def extract_openai_usage_from_chunk(
             usage["raw_usage"] = serialized
 
     elif provider_type == "responses":
-        # For Responses API, usage is only in chunk.response.usage for completed events
-        if hasattr(chunk, "type") and chunk.type == "response.completed":
+        # For Responses API, usage is only in chunk.response.usage on the terminal
+        # event. A run cut short (e.g. by max_output_tokens) ends on
+        # response.incomplete instead of response.completed, and is still billed.
+        if getattr(chunk, "type", None) in _RESPONSES_TERMINAL_EVENT_TYPES:
             if (
                 hasattr(chunk, "response")
                 and hasattr(chunk.response, "usage")
@@ -634,7 +645,8 @@ def extract_openai_content_from_chunk(
     Returns:
         For "chat": text content (str), or an audio/refusal delta block (dict),
         if present. For "responses": the full `response.output` list on the
-        `response.completed` event. None otherwise.
+        terminal (`response.completed`, `response.incomplete` or
+        `response.failed`) event. None otherwise.
     """
 
     if provider_type == "chat":
@@ -663,7 +675,7 @@ def extract_openai_content_from_chunk(
 
     elif provider_type == "responses":
         # Responses API format
-        if hasattr(chunk, "type") and chunk.type == "response.completed":
+        if getattr(chunk, "type", None) in _RESPONSES_TERMINAL_EVENT_TYPES:
             if hasattr(chunk, "response") and chunk.response:
                 res = chunk.response
                 if res.output:

posthog/test/ai/openai/test_openai.py

@@ -1916,6 +1916,85 @@ def test_streaming_responses_api_extracts_model_from_response_object(mock_client
         assert props["$ai_model"] == "gpt-4o-mini-stored"
 
 
+def test_streaming_responses_api_captures_usage_and_output_when_incomplete(
+    mock_client,
+):
+    """A stream cut short by max_output_tokens ends on response.incomplete, not
+    response.completed. Its usage and partial output are still billed and must be
+    captured."""
+    from openai.types.responses import ResponseIncompleteEvent
+    from openai.types.responses.response import IncompleteDetails
+
+    incomplete_response = Response(
+        id="resp_incomplete",
+        model="gpt-4o-mini",
+        object="response",
+        created_at=1741476542,
+        status="incomplete",
+        error=None,
+        incomplete_details=IncompleteDetails(reason="max_output_tokens"),
+        instructions=None,
+        max_output_tokens=16,
+        tools=[],
+        tool_choice="auto",
+        output=[
+            ResponseOutputMessage(
+                id="msg_123",
+                type="message",
+                role="assistant",
+                status="incomplete",
+                content=[
+                    ResponseOutputText(
+                        type="output_text",
+                        text="Once upon a time",
+                        annotations=[],
+                    )
+                ],
+            )
+        ],
+        parallel_tool_calls=True,
+        previous_response_id=None,
+        usage=make_response_usage(
+            input_tokens=20,
+            output_tokens=16,
+            total_tokens=36,
+        ),
+        user=None,
+        metadata={},
+    )
+    chunks = [
+        ResponseIncompleteEvent(
+            type="response.incomplete",
+            sequence_number=1,
+            response=incomplete_response,
+        )
+    ]
+
+    with patch("openai.resources.responses.Responses.create") as mock_create:
+        mock_create.return_value = iter(chunks)
+
+        client = OpenAI(api_key="test-key", posthog_client=mock_client)
+        response_generator = client.responses.create(
+            model="gpt-4o-mini",
+            input=[{"role": "user", "content": "Tell me a story"}],
+            max_output_tokens=16,
+            stream=True,
+            posthog_distinct_id="test-id",
+        )
+        list(response_generator)
+
+    props = mock_client.capture.call_args[1]["properties"]
+    assert props["$ai_stop_reason"] == "max_output_tokens"
+    assert props["$ai_input_tokens"] == 20
+    assert props["$ai_output_tokens"] == 16
+    assert props["$ai_output_choices"] == [
+        {
+            "role": "assistant",
+            "content": [{"type": "text", "text": "Once upon a time"}],
+        }
+    ]
+
+
 def test_non_streaming_extracts_model_from_response(mock_client):
     """Test that non-streaming calls extract model from response when not in kwargs."""
     # Create a response with model but we won't pass model in kwargs