fix(ai): only capture terminal, named Responses stop reasons

.sampo/changesets/responses-terminal-stop-reason.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: patch
+---
+
+Only terminal Responses API statuses become `$ai_stop_reason`: a queued or in-progress background run no longer records a lifecycle state as its stop reason, and an incomplete run is named by what cut it short (`incomplete_details.reason`, e.g. `max_output_tokens`). Streaming runs that end incomplete or failed now carry a stop reason too, and the LangChain callback reads stop reasons from `response_metadata` as well, covering Responses API and Anthropic runs that previously recorded none.

posthog/ai/langchain/callbacks.py

@@ -48,6 +48,7 @@
     _extract_cache_creation_ttl_breakdown,
     finalize_ai_content,
     get_model_params,
+    _responses_stop_reason,
     with_privacy_mode,
 )
 from posthog.client import Client
@@ -709,14 +710,11 @@ def _capture_generation(
                 finalize_ai_content(completions, self._ph_client),
             )
 
-            # Extract stop reason from generation info
+            # Extract the stop reason from the generation and its metadata
             if output.generations and output.generations[-1]:
-                last_gen = output.generations[-1][-1]
-                gen_info = getattr(last_gen, "generation_info", None)
-                if isinstance(gen_info, dict):
-                    finish_reason = gen_info.get("finish_reason")
-                    if finish_reason is not None:
-                        event_properties["$ai_stop_reason"] = finish_reason
+                stop_reason = _extract_stop_reason(output.generations[-1][-1])
+                if stop_reason is not None:
+                    event_properties["$ai_stop_reason"] = stop_reason
 
         _capture_ai_event(
             self._ph_client,
@@ -738,6 +736,37 @@ def _log_debug_event(
         )
 
 
+def _extract_stop_reason(generation: Any) -> Optional[str]:
+    """
+    Providers spread the stop reason across `generation_info` and
+    `response_metadata` under two spellings, so read every source in priority
+    order. The Responses API reports no finish_reason at all: an incomplete
+    run is named by what cut it short, and only terminal statuses count.
+    """
+
+    def as_dict(value: Any) -> dict:
+        return value if isinstance(value, dict) else {}
+
+    message = as_dict(
+        getattr(getattr(generation, "message", None), "response_metadata", None)
+    )
+    info = as_dict(getattr(generation, "generation_info", None))
+    nested = as_dict(info.get("response_metadata"))
+
+    for source, key in (
+        (message, "finish_reason"),
+        (message, "stop_reason"),
+        (info, "finish_reason"),
+        (nested, "stop_reason"),
+        (nested, "finish_reason"),
+        (info, "stop_reason"),
+    ):
+        if source.get(key) is not None:
+            return str(source[key])
+
+    return _responses_stop_reason(message) or _responses_stop_reason(nested)
+
+
 def _extract_raw_response(last_response):
     """Extract the response from the last response of the LLM call."""
     # We return the text of the response if not empty

posthog/ai/openai/_streaming.py

@@ -4,7 +4,7 @@
 from typing import Any, Dict, List, Optional
 
 from ..types import StreamingEventData, TokenUsage
-from ..utils import merge_usage_stats
+from ..utils import merge_usage_stats, _responses_stop_reason
 from .openai_converter import (
     accumulate_openai_tool_calls,
     extract_openai_content_from_chunk,
@@ -35,10 +35,12 @@ def process_chunk(self, chunk: Any) -> None:
         if content is not None:
             self.output.extend(content)
 
-        if getattr(chunk, "type", None) == "response.completed" and response:
-            status = getattr(response, "status", None)
-            if status is not None:
-                self.stop_reason = status
+        # A stream can end on response.completed, response.incomplete, or
+        # response.failed; any terminal response names the stop reason.
+        if response:
+            stop_reason = _responses_stop_reason(response)
+            if stop_reason is not None:
+                self.stop_reason = stop_reason
 
 
 @dataclass

posthog/ai/openai/openai_converter.py

@@ -17,7 +17,7 @@
     FormattedTextContent,
     TokenUsage,
 )
-from posthog.ai.utils import serialize_raw_usage
+from posthog.ai.utils import _responses_stop_reason, serialize_raw_usage
 
 
 def _item_attr(item: Any, name: str, default: Any = None) -> Any:
@@ -445,7 +445,7 @@ def extract_openai_stop_reason(response: Any) -> Optional[str]:
         return getattr(response.choices[0], "finish_reason", None)
     # Responses API
     if hasattr(response, "status"):
-        return getattr(response, "status", None)
+        return _responses_stop_reason(response)
     return None
 
 

posthog/ai/utils.py

@@ -303,6 +303,35 @@ def format_response(response, provider: str):
     return []
 
 
+# A Responses API run is only finished on these statuses; `queued` and
+# `in_progress` are lifecycle states a background run passes through.
+_TERMINAL_RESPONSE_STATUSES = frozenset(
+    {"completed", "failed", "cancelled", "incomplete"}
+)
+
+
+def _read_response_field(source: Any, key: str) -> Any:
+    return source.get(key) if isinstance(source, dict) else getattr(source, key, None)
+
+
+def _responses_stop_reason(response: Any) -> Optional[str]:
+    """
+    Map a Responses API outcome to a `$ai_stop_reason`: an incomplete run is
+    named by what cut it short (`incomplete_details.reason`, e.g.
+    `max_output_tokens`), the other terminal statuses stand for themselves,
+    and a non-terminal status yields None. Accepts an SDK response object or
+    a LangChain `response_metadata` dict.
+    """
+    status = _read_response_field(response, "status")
+    if not isinstance(status, str) or status not in _TERMINAL_RESPONSE_STATUSES:
+        return None
+    details = _read_response_field(response, "incomplete_details")
+    reason = _read_response_field(details, "reason")
+    if status == "incomplete" and isinstance(reason, str) and reason:
+        return reason
+    return status
+
+
 def extract_stop_reason(response: Any, provider: str) -> Optional[str]:
     """Extract stop reason from response based on provider."""
     if provider == "openai":

posthog/test/ai/langchain/test_callbacks.py

@@ -2839,3 +2839,58 @@ def test_ai_lane_client_routes_through_capture_ai(mock_client):
     events = [c[1]["event"] for c in mock_client.capture_ai.call_args_list]
     assert "$ai_generation" in events
     assert "$ai_trace" in events
+
+
+@pytest.mark.parametrize(
+    "generation_info,response_metadata,expected",
+    [
+        # generation_info finish_reason keeps priority
+        ({"finish_reason": "stop"}, {"status": "completed"}, "stop"),
+        # Responses API: a terminal status carries no finish_reason at all
+        (None, {"status": "completed", "incomplete_details": None}, "completed"),
+        # ... an incomplete run is named by what cut it short
+        (
+            None,
+            {
+                "status": "incomplete",
+                "incomplete_details": {"reason": "max_output_tokens"},
+            },
+            "max_output_tokens",
+        ),
+        # a queued background run has no stop reason yet
+        (None, {"status": "queued", "incomplete_details": None}, None),
+        # Anthropic reports through response_metadata.stop_reason
+        (None, {"stop_reason": "end_turn"}, "end_turn"),
+    ],
+)
+def test_stop_reason_resolution(
+    mock_client, generation_info, response_metadata, expected
+):
+    from langchain_core.outputs import ChatGeneration, LLMResult
+
+    cb = CallbackHandler(mock_client)
+    run_id = uuid.uuid4()
+    cb._set_llm_metadata(
+        serialized={},
+        run_id=run_id,
+        messages=[{"role": "user", "content": "test"}],
+        metadata={"ls_provider": "openai", "ls_model_name": "gpt-4o"},
+    )
+    response = LLMResult(
+        generations=[
+            [
+                ChatGeneration(
+                    message=AIMessage(
+                        content="Response", response_metadata=response_metadata
+                    ),
+                    generation_info=generation_info,
+                )
+            ]
+        ],
+        llm_output={},
+    )
+
+    cb._pop_run_and_capture_generation(run_id, None, response)
+
+    props = mock_client.capture.call_args.kwargs["properties"]
+    assert props.get("$ai_stop_reason") == expected

posthog/test/ai/openai/test_openai_converter.py

@@ -1,3 +1,5 @@
+import types
+
 import pytest
 
 try:
@@ -7,7 +9,11 @@
 except ImportError:
     OPENAI_AVAILABLE = False
 
-from posthog.ai.openai.openai_converter import format_openai_input
+from posthog.ai.openai._streaming import _ResponsesStreamState
+from posthog.ai.openai.openai_converter import (
+    extract_openai_stop_reason,
+    format_openai_input,
+)
 from posthog.test.ai.utils import make_response_usage
 
 pytestmark = pytest.mark.skipif(not OPENAI_AVAILABLE, reason="openai not available")
@@ -198,3 +204,56 @@ def _chunk(delta_kwargs):
 
     refusal_block = next(b for b in content if b["type"] == "refusal")
     assert refusal_block["refusal"] == "I can't help with that"
+
+
+def _response(status, incomplete_reason=None, **extra):
+    details = (
+        types.SimpleNamespace(reason=incomplete_reason) if incomplete_reason else None
+    )
+    return types.SimpleNamespace(status=status, incomplete_details=details, **extra)
+
+
+@pytest.mark.parametrize(
+    "status,incomplete_reason,expected",
+    [
+        # Terminal statuses stand for themselves
+        ("completed", None, "completed"),
+        ("failed", None, "failed"),
+        ("cancelled", None, "cancelled"),
+        # An incomplete run is named by what cut it short
+        ("incomplete", "max_output_tokens", "max_output_tokens"),
+        ("incomplete", None, "incomplete"),
+        # Lifecycle states of a background run are not stop reasons
+        ("queued", None, None),
+        ("in_progress", None, None),
+    ],
+)
+def test_extract_stop_reason_maps_responses_statuses(
+    status, incomplete_reason, expected
+):
+    assert extract_openai_stop_reason(_response(status, incomplete_reason)) == expected
+
+
+@pytest.mark.parametrize(
+    "status,incomplete_reason,expected",
+    [
+        ("completed", None, "completed"),
+        ("incomplete", "max_output_tokens", "max_output_tokens"),
+        ("failed", None, "failed"),
+        ("in_progress", None, None),
+    ],
+)
+def test_responses_stream_records_every_terminal_stop_reason(
+    status, incomplete_reason, expected
+):
+    state = _ResponsesStreamState()
+    state.process_chunk(
+        types.SimpleNamespace(
+            type=f"response.{status}",
+            response=_response(
+                status, incomplete_reason, model="gpt-4o", usage=None, output=[]
+            ),
+        )
+    )
+
+    assert state.stop_reason == expected