From c789b030ee2b7466d76cebc7816b1559161c0972 Mon Sep 17 00:00:00 2001 From: Bernat Torres Date: Thu, 3 Sep 2026 14:54:13 +0200 Subject: [PATCH 1/2] fix(ai): only capture terminal, named Responses stop reasons The Python SDK recorded raw Responses API statuses as $ai_stop_reason across three surfaces, none of which matched what the trace view understands. The native wrapper returned any status, so a background run recorded queued as its stop reason, and a truncated run recorded the bare incomplete instead of max_output_tokens. The streaming accumulator only read status on response.completed, so streams that ended incomplete or failed recorded nothing. The LangChain callback read only generation_info.finish_reason, so Responses API and Anthropic runs, which report through response_metadata, recorded nothing at all. One shared _responses_stop_reason helper now does the mapping: non-terminal statuses yield nothing, an incomplete run is named by incomplete_details.reason, and the other terminal statuses stand for themselves. All three surfaces route through it, and the LangChain callback reads the same metadata sources as the JS SDK's callback, in the same priority order. Ports PostHog/posthog-js#4700 and PostHog/posthog-js#4736 to Python. Generated-By: PostHog Desktop Task-Id: 5cb23cd6-224c-4d86-bc24-97730d4c2f98 --- .../responses-terminal-stop-reason.md | 5 ++ posthog/ai/langchain/callbacks.py | 49 ++++++++++-- posthog/ai/openai/_streaming.py | 12 +-- posthog/ai/openai/openai_converter.py | 4 +- posthog/ai/utils.py | 36 +++++++++ posthog/test/ai/langchain/test_callbacks.py | 58 ++++++++++++++ .../test/ai/openai/test_openai_converter.py | 80 +++++++++++++++++++ 7 files changed, 230 insertions(+), 14 deletions(-) create mode 100644 .sampo/changesets/responses-terminal-stop-reason.md diff --git a/.sampo/changesets/responses-terminal-stop-reason.md b/.sampo/changesets/responses-terminal-stop-reason.md new file mode 100644 index 000000000..cddddf2ef --- /dev/null +++ b/.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. diff --git a/posthog/ai/langchain/callbacks.py b/posthog/ai/langchain/callbacks.py index 91fdece52..6cb895828 100644 --- a/posthog/ai/langchain/callbacks.py +++ b/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,43 @@ 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. + """ + message = getattr(generation, "message", None) + message_metadata = getattr(message, "response_metadata", None) + if not isinstance(message_metadata, dict): + message_metadata = None + generation_info = getattr(generation, "generation_info", None) + if not isinstance(generation_info, dict): + generation_info = None + generation_metadata = ( + generation_info.get("response_metadata") if generation_info else None + ) + if not isinstance(generation_metadata, dict): + generation_metadata = None + + for source, key in ( + (message_metadata, "finish_reason"), + (message_metadata, "stop_reason"), + (generation_info, "finish_reason"), + (generation_metadata, "stop_reason"), + (generation_metadata, "finish_reason"), + (generation_info, "stop_reason"), + ): + value = source.get(key) if source else None + if value is not None: + return str(value) + + return _responses_stop_reason(message_metadata) or _responses_stop_reason( + generation_metadata + ) + + 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 diff --git a/posthog/ai/openai/_streaming.py b/posthog/ai/openai/_streaming.py index 3f2f38e74..23b6f5968 100644 --- a/posthog/ai/openai/_streaming.py +++ b/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 diff --git a/posthog/ai/openai/openai_converter.py b/posthog/ai/openai/openai_converter.py index ecd8c8c90..3f3bc869c 100644 --- a/posthog/ai/openai/openai_converter.py +++ b/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 diff --git a/posthog/ai/utils.py b/posthog/ai/utils.py index c456c0b72..523e1e844 100644 --- a/posthog/ai/utils.py +++ b/posthog/ai/utils.py @@ -303,6 +303,42 @@ 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: + if isinstance(source, dict): + return source.get(key) + return 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. + """ + if response is None: + return None + status = _read_response_field(response, "status") + if not isinstance(status, str) or status not in _TERMINAL_RESPONSE_STATUSES: + return None + if status == "incomplete": + details = _read_response_field(response, "incomplete_details") + reason = ( + _read_response_field(details, "reason") if details is not None else None + ) + if 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": diff --git a/posthog/test/ai/langchain/test_callbacks.py b/posthog/test/ai/langchain/test_callbacks.py index 11f021099..9dfeae61b 100644 --- a/posthog/test/ai/langchain/test_callbacks.py +++ b/posthog/test/ai/langchain/test_callbacks.py @@ -2839,3 +2839,61 @@ 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"] + if expected is None: + assert "$ai_stop_reason" not in props + else: + assert props["$ai_stop_reason"] == expected diff --git a/posthog/test/ai/openai/test_openai_converter.py b/posthog/test/ai/openai/test_openai_converter.py index 1c3668f68..c9c2a33c9 100644 --- a/posthog/test/ai/openai/test_openai_converter.py +++ b/posthog/test/ai/openai/test_openai_converter.py @@ -198,3 +198,83 @@ 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" + + +@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 +): + import types + + from posthog.ai.openai.openai_converter import extract_openai_stop_reason + + details = ( + types.SimpleNamespace(reason=incomplete_reason) if incomplete_reason else None + ) + response = types.SimpleNamespace(status=status, incomplete_details=details) + + assert extract_openai_stop_reason(response) == expected + + +def test_extract_stop_reason_keeps_chat_completions_passthrough(): + import types + + from posthog.ai.openai.openai_converter import extract_openai_stop_reason + + response = types.SimpleNamespace( + choices=[types.SimpleNamespace(finish_reason="stop")] + ) + + assert extract_openai_stop_reason(response) == "stop" + + +def _lifecycle_chunk(chunk_type, status, incomplete_reason=None): + import types + + details = ( + types.SimpleNamespace(reason=incomplete_reason) if incomplete_reason else None + ) + return types.SimpleNamespace( + type=chunk_type, + response=types.SimpleNamespace( + model="gpt-4o", + usage=None, + output=[], + status=status, + incomplete_details=details, + ), + ) + + +def test_responses_stream_records_stop_reason_for_every_terminal_event(): + from posthog.ai.openai._streaming import _ResponsesStreamState + + state = _ResponsesStreamState() + state.process_chunk(_lifecycle_chunk("response.in_progress", "in_progress")) + assert state.stop_reason is None + state.process_chunk( + _lifecycle_chunk("response.incomplete", "incomplete", "max_output_tokens") + ) + assert state.stop_reason == "max_output_tokens" + + failed = _ResponsesStreamState() + failed.process_chunk(_lifecycle_chunk("response.failed", "failed")) + assert failed.stop_reason == "failed" + + completed = _ResponsesStreamState() + completed.process_chunk(_lifecycle_chunk("response.completed", "completed")) + assert completed.stop_reason == "completed" From 36e4e269f6fe107dd910298d8fa8b3c50de94f8d Mon Sep 17 00:00:00 2001 From: Bernat Torres Date: Fri, 4 Sep 2026 14:37:47 +0200 Subject: [PATCH 2/2] chore(ai): tighten the stop reason helpers and their tests Same behavior, less of it. The metadata sources in the callback normalize to dicts once instead of each carrying its own isinstance guard, and the Responses helper folds its None guard and nested incomplete branch into the checks that already covered them. The tests drop a chat-completions passthrough case that test_openai.py already locks, hoist the imports the new cases share, and express the streaming behavior as a table rather than one test driving three accumulators, which also covers the failed status the old test missed. Generated-By: PostHog Desktop Task-Id: 5cb23cd6-224c-4d86-bc24-97730d4c2f98 --- posthog/ai/langchain/callbacks.py | 40 ++++----- posthog/ai/utils.py | 17 ++-- posthog/test/ai/langchain/test_callbacks.py | 5 +- .../test/ai/openai/test_openai_converter.py | 89 +++++++------------ 4 files changed, 57 insertions(+), 94 deletions(-) diff --git a/posthog/ai/langchain/callbacks.py b/posthog/ai/langchain/callbacks.py index 6cb895828..49832a05d 100644 --- a/posthog/ai/langchain/callbacks.py +++ b/posthog/ai/langchain/callbacks.py @@ -743,34 +743,28 @@ def _extract_stop_reason(generation: Any) -> Optional[str]: 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. """ - message = getattr(generation, "message", None) - message_metadata = getattr(message, "response_metadata", None) - if not isinstance(message_metadata, dict): - message_metadata = None - generation_info = getattr(generation, "generation_info", None) - if not isinstance(generation_info, dict): - generation_info = None - generation_metadata = ( - generation_info.get("response_metadata") if generation_info else None + + def as_dict(value: Any) -> dict: + return value if isinstance(value, dict) else {} + + message = as_dict( + getattr(getattr(generation, "message", None), "response_metadata", None) ) - if not isinstance(generation_metadata, dict): - generation_metadata = None + info = as_dict(getattr(generation, "generation_info", None)) + nested = as_dict(info.get("response_metadata")) for source, key in ( - (message_metadata, "finish_reason"), - (message_metadata, "stop_reason"), - (generation_info, "finish_reason"), - (generation_metadata, "stop_reason"), - (generation_metadata, "finish_reason"), - (generation_info, "stop_reason"), + (message, "finish_reason"), + (message, "stop_reason"), + (info, "finish_reason"), + (nested, "stop_reason"), + (nested, "finish_reason"), + (info, "stop_reason"), ): - value = source.get(key) if source else None - if value is not None: - return str(value) + if source.get(key) is not None: + return str(source[key]) - return _responses_stop_reason(message_metadata) or _responses_stop_reason( - generation_metadata - ) + return _responses_stop_reason(message) or _responses_stop_reason(nested) def _extract_raw_response(last_response): diff --git a/posthog/ai/utils.py b/posthog/ai/utils.py index 523e1e844..a83ace709 100644 --- a/posthog/ai/utils.py +++ b/posthog/ai/utils.py @@ -311,9 +311,7 @@ def format_response(response, provider: str): def _read_response_field(source: Any, key: str) -> Any: - if isinstance(source, dict): - return source.get(key) - return getattr(source, key, None) + return source.get(key) if isinstance(source, dict) else getattr(source, key, None) def _responses_stop_reason(response: Any) -> Optional[str]: @@ -324,18 +322,13 @@ def _responses_stop_reason(response: Any) -> Optional[str]: and a non-terminal status yields None. Accepts an SDK response object or a LangChain `response_metadata` dict. """ - if response is None: - return None status = _read_response_field(response, "status") if not isinstance(status, str) or status not in _TERMINAL_RESPONSE_STATUSES: return None - if status == "incomplete": - details = _read_response_field(response, "incomplete_details") - reason = ( - _read_response_field(details, "reason") if details is not None else None - ) - if isinstance(reason, str) and reason: - return reason + 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 diff --git a/posthog/test/ai/langchain/test_callbacks.py b/posthog/test/ai/langchain/test_callbacks.py index 9dfeae61b..43325ac05 100644 --- a/posthog/test/ai/langchain/test_callbacks.py +++ b/posthog/test/ai/langchain/test_callbacks.py @@ -2893,7 +2893,4 @@ def test_stop_reason_resolution( cb._pop_run_and_capture_generation(run_id, None, response) props = mock_client.capture.call_args.kwargs["properties"] - if expected is None: - assert "$ai_stop_reason" not in props - else: - assert props["$ai_stop_reason"] == expected + assert props.get("$ai_stop_reason") == expected diff --git a/posthog/test/ai/openai/test_openai_converter.py b/posthog/test/ai/openai/test_openai_converter.py index c9c2a33c9..e23ecac87 100644 --- a/posthog/test/ai/openai/test_openai_converter.py +++ b/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") @@ -200,6 +206,13 @@ def _chunk(delta_kwargs): 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", [ @@ -218,63 +231,29 @@ def _chunk(delta_kwargs): def test_extract_stop_reason_maps_responses_statuses( status, incomplete_reason, expected ): - import types - - from posthog.ai.openai.openai_converter import extract_openai_stop_reason + assert extract_openai_stop_reason(_response(status, incomplete_reason)) == expected - details = ( - types.SimpleNamespace(reason=incomplete_reason) if incomplete_reason else None - ) - response = types.SimpleNamespace(status=status, incomplete_details=details) - - assert extract_openai_stop_reason(response) == expected - - -def test_extract_stop_reason_keeps_chat_completions_passthrough(): - import types - - from posthog.ai.openai.openai_converter import extract_openai_stop_reason - - response = types.SimpleNamespace( - choices=[types.SimpleNamespace(finish_reason="stop")] - ) - - assert extract_openai_stop_reason(response) == "stop" - - -def _lifecycle_chunk(chunk_type, status, incomplete_reason=None): - import types - - details = ( - types.SimpleNamespace(reason=incomplete_reason) if incomplete_reason else None - ) - return types.SimpleNamespace( - type=chunk_type, - response=types.SimpleNamespace( - model="gpt-4o", - usage=None, - output=[], - status=status, - incomplete_details=details, - ), - ) - - -def test_responses_stream_records_stop_reason_for_every_terminal_event(): - from posthog.ai.openai._streaming import _ResponsesStreamState +@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(_lifecycle_chunk("response.in_progress", "in_progress")) - assert state.stop_reason is None state.process_chunk( - _lifecycle_chunk("response.incomplete", "incomplete", "max_output_tokens") + types.SimpleNamespace( + type=f"response.{status}", + response=_response( + status, incomplete_reason, model="gpt-4o", usage=None, output=[] + ), + ) ) - assert state.stop_reason == "max_output_tokens" - - failed = _ResponsesStreamState() - failed.process_chunk(_lifecycle_chunk("response.failed", "failed")) - assert failed.stop_reason == "failed" - completed = _ResponsesStreamState() - completed.process_chunk(_lifecycle_chunk("response.completed", "completed")) - assert completed.stop_reason == "completed" + assert state.stop_reason == expected