Skip to content

fix(ai): capture an error event when a streaming call fails to connect #1146

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

Open
pcbeingused333 wants to merge 3 commits into
base: main
Choose a base branch
from
+307 −12
Open

fix(ai): capture an error event when a streaming call fails to connect #1146

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7aaa93c
fix(ai): capture an error event when a streaming call fails to connect
pcbeingused333 Oct 29, 2026
29325a4
fix(ai): annotate the Gemini streaming accumulator for mypy
pcbeingused333 Oct 30, 2026
9798de2
Merge remote-tracking branch 'origin/main' into fix/stream-connect-er…
pcbeingused333 Oct 30, 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/stream-connect-error.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
pypi/posthog: patch
---

Capture an `$ai_generation` error event when `create(stream=True)` fails before the stream starts (bad key, rate limit, invalid model) in the OpenAI chat, OpenAI Responses, Anthropic and Gemini wrappers, sync and async. Previously only errors raised mid-stream were captured.
19 changes: 18 additions & 1 deletion posthog/ai/anthropic/anthropic.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -162,7 +162,24 @@ def _create_streaming(
**kwargs: Any,
):
start_time = time.time()
response = super().create(**kwargs)
try:
response = super().create(**kwargs)
except Exception as exc:
accumulator = _AnthropicStreamAccumulator()
self._capture_streaming_event(

@marandaneto marandaneto Oct 29, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

blocking: Preserve the provider exception when error telemetry cannot format rejected input — When Anthropic rejects malformed input such as messages=[None] with HTTP 400, the newly added capture call formats that same rejected input before reaching the bare raise. format_anthropic_input calls msg.get(...), raising AttributeError and replacing the original anthropic.BadRequestError; the async handler at posthog/ai/anthropic/anthropic_async.py:169 behaves identically. Previously callers received the actionable provider error and could handle BadRequestError normally. Ensure failures in this telemetry path cannot replace the original exception. Reproduction: reproduced — python -m pytest -p no:cacheprovider posthog/test/ai/test_review_connect.py -k invalid_anthropic --timeout=30 -vv fails both sync/async assertions on the reviewed head with AttributeError: 'NoneType' object has no attribute 'get', while both pass unchanged at the recorded merge base.

posthog_distinct_id,
posthog_trace_id,
posthog_properties,
posthog_privacy_mode,
posthog_groups,
kwargs,
accumulator.usage_stats,
time.time() - start_time,
accumulator.content_blocks,
accumulator.accumulated_content,
error=exc,
)
raise
return self._track_streaming_response(
response,
posthog_distinct_id,
Expand Down
19 changes: 18 additions & 1 deletion posthog/ai/anthropic/anthropic_async.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -162,7 +162,24 @@ async def _create_streaming(
**kwargs: Any,
):
start_time = time.time()
response = await super().create(**kwargs)
try:
response = await super().create(**kwargs)
except Exception as exc:
accumulator = _AnthropicStreamAccumulator()
await self._capture_streaming_event(
posthog_distinct_id,
posthog_trace_id,
posthog_properties,
posthog_privacy_mode,
posthog_groups,
kwargs,
accumulator.usage_stats,
time.time() - start_time,
accumulator.content_blocks,
accumulator.accumulated_content,
error=exc,
)
raise
return self._track_streaming_response(
response,
posthog_distinct_id,
Expand Down
24 changes: 22 additions & 2 deletions posthog/ai/gemini/gemini.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -256,11 +256,31 @@ def _generate_content_streaming(
):
start_time = time.time()
usage_stats: TokenUsage = TokenUsage()
accumulated_content = []
accumulated_content: list[Any] = []
stop_reason: Optional[str] = None

kwargs_without_stream = {"model": model, "contents": contents, **kwargs}
response = self._client.models.generate_content_stream(**kwargs_without_stream)
try:
response = self._client.models.generate_content_stream(
**kwargs_without_stream
)
except Exception as exc:
self._capture_streaming_event(
model,
contents,
distinct_id,
trace_id,
properties,
privacy_mode,
groups,
kwargs,
usage_stats,
time.time() - start_time,
accumulated_content,

@marandaneto marandaneto Oct 29, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

blocking: Keep the new Gemini handlers compatible with Mypy — Passing the initially empty, unannotated accumulated_content into this new exception handler makes Mypy require an explicit element type before the nested generator populates it. Both the sync and async additions introduce var-annotated errors and fail the required Mypy check; please annotate the accumulated content in both wrappers. Reproduction: reproduced — python -m mypy --no-site-packages --config-file mypy.ini posthog/ai/gemini/gemini.py posthog/ai/gemini/gemini_async.py reports exactly two annotation errors on the reviewed head and no issues on the recorded merge base.

stop_reason=stop_reason,
error=exc,
)
raise

def generator():
nonlocal usage_stats
Expand Down
26 changes: 22 additions & 4 deletions posthog/ai/gemini/gemini_async.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -240,13 +240,31 @@ async def _generate_content_streaming(
):
start_time = time.time()
usage_stats: TokenUsage = TokenUsage()
accumulated_content = []
accumulated_content: list[Any] = []
stop_reason: Optional[str] = None

kwargs_without_stream = {"model": model, "contents": contents, **kwargs}
response = await self._client.aio.models.generate_content_stream(
**kwargs_without_stream
)
try:
response = await self._client.aio.models.generate_content_stream(
**kwargs_without_stream
)
except Exception as exc:
self._capture_streaming_event(
model,
contents,
distinct_id,
trace_id,
properties,
privacy_mode,
groups,
kwargs,
usage_stats,
time.time() - start_time,
accumulated_content,
stop_reason=stop_reason,
error=exc,
)
raise

async def async_generator():
nonlocal usage_stats
Expand Down
32 changes: 30 additions & 2 deletions posthog/ai/openai/openai.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,7 +167,21 @@ def _create_streaming(
):
start_time = time.time()
state = _ResponsesStreamState()
response = self._original.create(**kwargs)
try:
response = self._original.create(**kwargs)
except Exception as exc:
self._capture_streaming_event(
posthog_distinct_id,
posthog_trace_id,
posthog_properties,
posthog_privacy_mode,
posthog_groups,
kwargs,
state,
time.time() - start_time,
error=exc,
)
raise

def generator():
stream_error: Optional[BaseException] = None
Expand Down Expand Up @@ -395,7 +409,21 @@ def _create_streaming(
**(stream_options if isinstance(stream_options, dict) else {}),
"include_usage": True,
}
response = self._original.create(**kwargs)
try:
response = self._original.create(**kwargs)
except Exception as exc:
self._capture_streaming_event(
posthog_distinct_id,
posthog_trace_id,
posthog_properties,
posthog_privacy_mode,
posthog_groups,
kwargs,
state,
time.time() - start_time,
error=exc,
)
raise

def generator():
stream_error: Optional[BaseException] = None
Expand Down
32 changes: 30 additions & 2 deletions posthog/ai/openai/openai_async.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -168,7 +168,21 @@ async def _create_streaming(
):
start_time = time.time()
state = _ResponsesStreamState()
response = await self._original.create(**kwargs)
try:
response = await self._original.create(**kwargs)
except Exception as exc:
await self._capture_streaming_event(
posthog_distinct_id,
posthog_trace_id,
posthog_properties,
posthog_privacy_mode,
posthog_groups,
kwargs,
state,
time.time() - start_time,
error=exc,
)
raise

async def async_generator():
stream_error: Optional[BaseException] = None
Expand Down Expand Up @@ -399,7 +413,21 @@ async def _create_streaming(
**(stream_options if isinstance(stream_options, dict) else {}),
"include_usage": True,
}
response = await self._original.create(**kwargs)
try:
response = await self._original.create(**kwargs)
except Exception as exc:
await self._capture_streaming_event(
posthog_distinct_id,
posthog_trace_id,
posthog_properties,
posthog_privacy_mode,
posthog_groups,
kwargs,
state,
time.time() - start_time,
error=exc,
)
raise

async def async_generator():
stream_error: Optional[BaseException] = None
Expand Down
162 changes: 162 additions & 0 deletions posthog/test/ai/test_stream_midstream_error.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -267,3 +267,165 @@ async def stream():
props = _props(mock_client)
assert props["$ai_is_error"] is True
assert props["$ai_http_status"] == 0


class _ConnectError(RuntimeError):
status_code = 401


def test_openai_chat_stream_connect_error(mock_client):
from posthog.ai.openai import OpenAI

with patch("openai.resources.chat.completions.Completions.create") as create:
create.side_effect = _ConnectError("bad key")
client = OpenAI(api_key="k", posthog_client=mock_client)
with pytest.raises(_ConnectError):
client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": "hi"}],
stream=True,
)

props = _props(mock_client)
assert props["$ai_is_error"] is True
assert props["$ai_http_status"] == 401


@pytest.mark.asyncio
async def test_openai_chat_stream_connect_error_async(mock_client):
from posthog.ai.openai import AsyncOpenAI

async def create(**kwargs):
raise _ConnectError("bad key")

with patch(
"openai.resources.chat.completions.AsyncCompletions.create", side_effect=create
):
client = AsyncOpenAI(api_key="k", posthog_client=mock_client)
with pytest.raises(_ConnectError):
await client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": "hi"}],
stream=True,
)

props = _props(mock_client)
assert props["$ai_is_error"] is True
assert props["$ai_http_status"] == 401


def test_openai_responses_stream_connect_error(mock_client):
from posthog.ai.openai import OpenAI

with patch("openai.resources.responses.Responses.create") as create:
create.side_effect = _ConnectError("bad key")
client = OpenAI(api_key="k", posthog_client=mock_client)
with pytest.raises(_ConnectError):
client.responses.create(model="gpt-4", input="hi", stream=True)

props = _props(mock_client)
assert props["$ai_is_error"] is True
assert props["$ai_http_status"] == 401


@pytest.mark.asyncio
async def test_openai_responses_stream_connect_error_async(mock_client):
from posthog.ai.openai import AsyncOpenAI

async def create(**kwargs):
raise _ConnectError("bad key")

with patch("openai.resources.responses.AsyncResponses.create", side_effect=create):
client = AsyncOpenAI(api_key="k", posthog_client=mock_client)
with pytest.raises(_ConnectError):
await client.responses.create(model="gpt-4", input="hi", stream=True)

props = _props(mock_client)
assert props["$ai_is_error"] is True
assert props["$ai_http_status"] == 401


def test_anthropic_stream_connect_error(mock_client):
from posthog.ai.anthropic import Anthropic

with patch("anthropic.resources.messages.Messages.create") as create:
create.side_effect = _ConnectError("bad key")
client = Anthropic(api_key="k", posthog_client=mock_client)
with pytest.raises(_ConnectError):
client.messages.create(
model="claude-3-opus-20240229",
messages=[{"role": "user", "content": "hi"}],
max_tokens=5,
stream=True,
)

props = _props(mock_client)
assert props["$ai_is_error"] is True
assert props["$ai_http_status"] == 401


@pytest.mark.asyncio
async def test_anthropic_stream_connect_error_async(mock_client):
from posthog.ai.anthropic import AsyncAnthropic

async def create(**kwargs):
raise _ConnectError("bad key")

with patch("anthropic.resources.messages.AsyncMessages.create", side_effect=create):
client = AsyncAnthropic(api_key="k", posthog_client=mock_client)
with pytest.raises(_ConnectError):
await client.messages.create(
model="claude-3-opus-20240229",
messages=[{"role": "user", "content": "hi"}],
max_tokens=5,
stream=True,
)

props = _props(mock_client)
assert props["$ai_is_error"] is True
assert props["$ai_http_status"] == 401


def test_gemini_stream_connect_error(mock_client):
pytest.importorskip("google.genai")
from google import genai as google_genai

from posthog.ai.gemini import Client

with patch.object(google_genai, "Client") as cls:
inst = MagicMock()
cls.return_value = inst
inst.models.generate_content_stream.side_effect = _ConnectError("bad key")
client = Client(api_key="k", posthog_client=mock_client)
with pytest.raises(_ConnectError):
client.models.generate_content_stream(
model="gemini-2.0-flash", contents=["hi"]
)

props = _props(mock_client)
assert props["$ai_is_error"] is True
assert props["$ai_http_status"] == 401


@pytest.mark.asyncio
async def test_gemini_stream_connect_error_async(mock_client):
pytest.importorskip("google.genai")
from google import genai as google_genai

from posthog.ai.gemini import AsyncClient

with patch.object(google_genai, "Client") as cls:
inst = MagicMock()
cls.return_value = inst
inst.aio.models.generate_content_stream = AsyncMock(
side_effect=_ConnectError("bad key")
)
client = AsyncClient(api_key="k", posthog_client=mock_client)
with pytest.raises(_ConnectError):
await client.models.generate_content_stream(
model="gemini-2.0-flash", contents=["hi"]
)

props = _props(mock_client)
assert props["$ai_is_error"] is True
assert props["$ai_http_status"] == 401