Skip to content

fix(ai): record mid-stream errors on streamed OpenAI, Anthropic and Gemini events #1094

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
marandaneto merged 2 commits into from
Oct 22, 2026
Merged

fix(ai): record mid-stream errors on streamed OpenAI, Anthropic and Gemini events #1094

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bc83070
fix(ai): record mid-stream errors on streamed OpenAI, Anthropic and G…
pcbeingused333 Oct 21, 2026
492f2ca
fix(ai): pass the stream error to StreamingEventData directly and upd…
pcbeingused333 Oct 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .sampo/changesets/streaming-midstream-error.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
pypi/posthog: patch
---

Streamed OpenAI, Anthropic and Gemini generations that raise mid-stream are now captured with `$ai_is_error` and `$ai_error` (and the error's status code as `$ai_http_status`) instead of being recorded as a successful 200. The error is still re-raised to the caller, matching what non-streaming calls already did.
7 changes: 7 additions & 0 deletions posthog/ai/anthropic/anthropic.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -188,10 +188,14 @@ def _track_streaming_response(
accumulator = _AnthropicStreamAccumulator()

def generator():
stream_error: Optional[BaseException] = None
try:
for event in response:
accumulator.consume(event)
yield event
except Exception as exc:
stream_error = exc
raise
finally:
end_time = time.time()
latency = end_time - start_time
Expand All @@ -208,6 +212,7 @@ def generator():
accumulator.content_blocks,
accumulator.accumulated_content,
stop_reason=accumulator.stop_reason,
error=stream_error,
)

return _StreamWrapper(generator(), stream=response)
Expand All @@ -225,6 +230,7 @@ def _capture_streaming_event(
content_blocks: List[StreamingContentBlock],
accumulated_content: str,
stop_reason: Optional[str] = None,
error: Optional[BaseException] = None,
):
from posthog.ai.types import StreamingEventData
from posthog.ai.anthropic.anthropic_converter import (
Expand Down Expand Up @@ -252,6 +258,7 @@ def _capture_streaming_event(
privacy_mode=posthog_privacy_mode,
groups=posthog_groups,
stop_reason=stop_reason,
error=error,
)

# Use the common capture function
Expand Down
7 changes: 7 additions & 0 deletions posthog/ai/anthropic/anthropic_async.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -188,10 +188,14 @@ def _track_streaming_response(
accumulator = _AnthropicStreamAccumulator()

async def generator():
stream_error: Optional[BaseException] = None
try:
async for event in response:
accumulator.consume(event)
yield event
except Exception as exc:
stream_error = exc
raise
finally:
end_time = time.time()
latency = end_time - start_time
Expand All @@ -208,6 +212,7 @@ async def generator():
accumulator.content_blocks,
accumulator.accumulated_content,
stop_reason=accumulator.stop_reason,
error=stream_error,
)

return AsyncStreamWrapper(generator(), stream=response)
Expand All @@ -225,6 +230,7 @@ async def _capture_streaming_event(
content_blocks: List[StreamingContentBlock],
accumulated_content: str,
stop_reason: Optional[str] = None,
error: Optional[BaseException] = None,
):
from posthog.ai.types import StreamingEventData
from posthog.ai.anthropic.anthropic_converter import (
Expand Down Expand Up @@ -252,6 +258,7 @@ async def _capture_streaming_event(
privacy_mode=posthog_privacy_mode,
groups=posthog_groups,
stop_reason=stop_reason,
error=error,
)

# Use the common capture function
Expand Down
2 changes: 2 additions & 0 deletions posthog/ai/gemini/_shared.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -203,6 +203,7 @@ def _capture_streaming_event(
latency: float,
output: Any,
stop_reason: Optional[str] = None,
error: Optional[BaseException] = None,
) -> None:
formatted_input = merge_system_prompt(
{"contents": contents, **kwargs}, "gemini"
Expand All @@ -222,6 +223,7 @@ def _capture_streaming_event(
privacy_mode=privacy_mode,
groups=groups,
stop_reason=stop_reason,
error=error,
)
capture_streaming_event(self._ph_client, event_data)

Expand Down
5 changes: 5 additions & 0 deletions posthog/ai/gemini/gemini.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -266,6 +266,7 @@ def generator():
nonlocal usage_stats
nonlocal accumulated_content
nonlocal stop_reason
stream_error: Optional[BaseException] = None
try:
for chunk in response:
# Extract usage stats from chunk
Expand All @@ -288,6 +289,9 @@ def generator():

yield chunk

except Exception as exc:
stream_error = exc
raise
finally:
end_time = time.time()
latency = end_time - start_time
Expand All @@ -305,6 +309,7 @@ def generator():
latency,
accumulated_content,
stop_reason=stop_reason,
error=stream_error,
)

return generator()
Expand Down
5 changes: 5 additions & 0 deletions posthog/ai/gemini/gemini_async.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -253,6 +253,7 @@ async def async_generator():
nonlocal accumulated_content
nonlocal stop_reason

stream_error: Optional[BaseException] = None
try:
async for chunk in response:
# Extract usage stats from chunk
Expand All @@ -275,6 +276,9 @@ async def async_generator():

yield chunk

except Exception as exc:
stream_error = exc
raise
finally:
end_time = time.time()
latency = end_time - start_time
Expand All @@ -292,6 +296,7 @@ async def async_generator():
latency,
accumulated_content,
stop_reason=stop_reason,
error=stream_error,
)

return AsyncStreamWrapper(async_generator(), stream=response)
Expand Down
2 changes: 2 additions & 0 deletions posthog/ai/openai/_streaming.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,7 @@ def _build_streaming_event_data(
model_from_response: Optional[str],
stop_reason: Optional[str],
service_tier: Optional[str] = None,
error: Optional[BaseException] = None,
) -> StreamingEventData:
"""Build the fields shared by both OpenAI streaming endpoint events."""

Expand All @@ -121,4 +122,5 @@ def _build_streaming_event_data(
groups=groups,
stop_reason=stop_reason,
service_tier=service_tier,
error=error,
)
14 changes: 14 additions & 0 deletions posthog/ai/openai/openai.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,10 +170,14 @@ def _create_streaming(
response = self._original.create(**kwargs)

def generator():
stream_error: Optional[BaseException] = None
try:
for chunk in response:
state.process_chunk(chunk)
yield chunk
except Exception as exc:
stream_error = exc
raise
finally:
self._capture_streaming_event(
posthog_distinct_id,
Expand All @@ -184,6 +188,7 @@ def generator():
kwargs,
state,
time.time() - start_time,
error=stream_error,
)

return _StreamWrapper(generator(), stream=response)
Expand All @@ -198,6 +203,7 @@ def _capture_streaming_event(
kwargs: Dict[str, Any],
state: _ResponsesStreamState,
latency: float,
error: Optional[BaseException] = None,
):
from posthog.ai.utils import capture_streaming_event

Expand All @@ -216,6 +222,7 @@ def _capture_streaming_event(
model_from_response=state.model,
stop_reason=state.stop_reason,
service_tier=state.service_tier,
error=error,
)
capture_streaming_event(self._client._ph_client, event_data)

Expand Down Expand Up @@ -388,10 +395,14 @@ def _create_streaming(
response = self._original.create(**kwargs)

def generator():
stream_error: Optional[BaseException] = None
try:
for chunk in response:
state.process_chunk(chunk)
yield chunk
except Exception as exc:
stream_error = exc
raise
finally:
self._capture_streaming_event(
posthog_distinct_id,
Expand All @@ -402,6 +413,7 @@ def generator():
kwargs,
state,
time.time() - start_time,
error=stream_error,
)

return _StreamWrapper(generator(), stream=response)
Expand All @@ -416,6 +428,7 @@ def _capture_streaming_event(
kwargs: Dict[str, Any],
state: _ChatCompletionsStreamState,
latency: float,
error: Optional[BaseException] = None,
):
from posthog.ai.utils import capture_streaming_event

Expand All @@ -436,6 +449,7 @@ def _capture_streaming_event(
model_from_response=state.model,
stop_reason=state.stop_reason,
service_tier=state.service_tier,
error=error,
)
capture_streaming_event(self._client._ph_client, event_data)

Expand Down
14 changes: 14 additions & 0 deletions posthog/ai/openai/openai_async.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -171,10 +171,14 @@ async def _create_streaming(
response = await self._original.create(**kwargs)

async def async_generator():
stream_error: Optional[BaseException] = None
try:
async for chunk in response:
state.process_chunk(chunk)
yield chunk
except Exception as exc:
stream_error = exc
raise
finally:
await self._capture_streaming_event(
posthog_distinct_id,
Expand All @@ -185,6 +189,7 @@ async def async_generator():
kwargs,
state,
time.time() - start_time,
error=stream_error,
)

return AsyncStreamWrapper(async_generator(), stream=response)
Expand All @@ -199,6 +204,7 @@ async def _capture_streaming_event(
kwargs: Dict[str, Any],
state: _ResponsesStreamState,
latency: float,
error: Optional[BaseException] = None,
):
from posthog.ai.utils import capture_streaming_event

Expand All @@ -217,6 +223,7 @@ async def _capture_streaming_event(
model_from_response=state.model,
stop_reason=state.stop_reason,
service_tier=state.service_tier,
error=error,
)
capture_streaming_event(self._client._ph_client, event_data)

Expand Down Expand Up @@ -392,10 +399,14 @@ async def _create_streaming(
response = await self._original.create(**kwargs)

async def async_generator():
stream_error: Optional[BaseException] = None
try:
async for chunk in response:
state.process_chunk(chunk)
yield chunk
except Exception as exc:
stream_error = exc
raise
finally:
await self._capture_streaming_event(
posthog_distinct_id,
Expand All @@ -406,6 +417,7 @@ async def async_generator():
kwargs,
state,
time.time() - start_time,
error=stream_error,
)

return AsyncStreamWrapper(async_generator(), stream=response)
Expand All @@ -420,6 +432,7 @@ async def _capture_streaming_event(
kwargs: Dict[str, Any],
state: _ChatCompletionsStreamState,
latency: float,
error: Optional[BaseException] = None,
):
from posthog.ai.utils import capture_streaming_event

Expand All @@ -440,6 +453,7 @@ async def _capture_streaming_event(
model_from_response=state.model,
stop_reason=state.stop_reason,
service_tier=state.service_tier,
error=error,
)
capture_streaming_event(self._client._ph_client, event_data)

Expand Down
1 change: 1 addition & 0 deletions posthog/ai/types.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -143,3 +143,4 @@ class StreamingEventData(TypedDict):
groups: Optional[Dict[str, Any]]
stop_reason: Optional[str]
service_tier: NotRequired[Optional[str]]
error: NotRequired[Optional[BaseException]] # raised mid-stream, if any

@dustinbyrne dustinbyrne Oct 21, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

F2 — [P2] Refresh the generated public API snapshot

MINOR · OPTIONAL acceptance feedback; outstanding mechanical merge prerequisite · Introduced reference drift
Location: posthog/ai/types.py:146, RIGHT.

Adding the optional StreamingEventData.error member changes the generated API surface, but references/public_api_snapshot.txt still describes the base type.

The snapshot checker passes on base bf873523c700d9940c8ec86b677d2bd4904dfa55 and exits 1 on head ec98ed6794f2a866b0ae40522b05dfff8b878024. Its only generated difference is:

+attribute posthog.ai.types.StreamingEventData.error: NotRequired[Optional[BaseException]]

CONTRIBUTING.md:33–43 requires regenerating the reference for API-surface changes, and .github/workflows/ci.yml:80–95 runs the snapshot check.

This is a generated-reference/check failure, not a demonstrated runtime compatibility break: the member is optional and existing constructions remain valid. Consider regenerating and including the reference update, then checking it before merge:

uv run --no-sync make public_api_snapshot
uv run --no-sync make public_api_check

Verification: Executed .github/scripts/check_public_api.py without --write on both revisions under equivalent Python 3.11.13/griffe 2.0.2 environments. The optional member is the sole difference. This is deterministic artifact validation, not a behavioral test. No regeneration was performed.

8 changes: 7 additions & 1 deletion posthog/ai/utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -837,6 +837,7 @@ def capture_streaming_event(
event_data: Standardized streaming event data containing all necessary information
"""
trace_id = event_data.get("trace_id") or str(uuid.uuid4())
error = event_data.get("error")

# Omitted when the provider never reported a count: absent means unknown,
# 0 is a report of nothing. An interrupted stream often reports neither.
Expand Down Expand Up @@ -865,13 +866,18 @@ def capture_streaming_event(
event_data["privacy_mode"],
finalize_ai_content(event_data["formatted_output"], ph_client),
),
"$ai_http_status": 200,
"$ai_http_status": (
getattr(error, "status_code", 0) if error is not None else 200
),
**({"$ai_input_tokens": input_tokens} if input_tokens is not None else {}),
**({"$ai_output_tokens": output_tokens} if output_tokens is not None else {}),
"$ai_latency": event_data["latency"],
"$ai_trace_id": trace_id,
"$ai_base_url": str(event_data["base_url"]),
**(event_data.get("properties") or {}),
**(
{"$ai_is_error": True, "$ai_error": str(error)} if error is not None else {}
),
}

warn_if_posthog_ai_gateway(event_data["base_url"])
Expand Down
Loading
Loading