From 7aaa93c9203f85e1426704bc62037395cab957f9 Mon Sep 17 00:00:00 2001 From: Alex Castillo Date: Thu, 29 Oct 2026 06:32:57 -0400 Subject: [PATCH 1/4] fix(ai): capture an error event when a streaming call fails to connect Co-Authored-By: Claude Sonnet 5.5 Signed-off-by: Alex Castillo --- .sampo/changesets/stream-connect-error.md | 5 + posthog/ai/anthropic/anthropic.py | 19 +- posthog/ai/anthropic/anthropic_async.py | 19 +- posthog/ai/gemini/gemini.py | 22 ++- posthog/ai/gemini/gemini_async.py | 24 ++- posthog/ai/openai/openai.py | 32 +++- posthog/ai/openai/openai_async.py | 32 +++- .../test/ai/test_stream_midstream_error.py | 162 ++++++++++++++++++ 8 files changed, 305 insertions(+), 10 deletions(-) create mode 100644 .sampo/changesets/stream-connect-error.md diff --git a/.sampo/changesets/stream-connect-error.md b/.sampo/changesets/stream-connect-error.md new file mode 100644 index 000000000..9f684067b --- /dev/null +++ b/.sampo/changesets/stream-connect-error.md @@ -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. diff --git a/posthog/ai/anthropic/anthropic.py b/posthog/ai/anthropic/anthropic.py index 704cacb20..0267eb84f 100644 --- a/posthog/ai/anthropic/anthropic.py +++ b/posthog/ai/anthropic/anthropic.py @@ -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( + 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, diff --git a/posthog/ai/anthropic/anthropic_async.py b/posthog/ai/anthropic/anthropic_async.py index 5207fa6b4..ce08ad0f2 100644 --- a/posthog/ai/anthropic/anthropic_async.py +++ b/posthog/ai/anthropic/anthropic_async.py @@ -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, diff --git a/posthog/ai/gemini/gemini.py b/posthog/ai/gemini/gemini.py index dad87da9a..cacf8e4fc 100644 --- a/posthog/ai/gemini/gemini.py +++ b/posthog/ai/gemini/gemini.py @@ -260,7 +260,27 @@ def _generate_content_streaming( 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, + stop_reason=stop_reason, + error=exc, + ) + raise def generator(): nonlocal usage_stats diff --git a/posthog/ai/gemini/gemini_async.py b/posthog/ai/gemini/gemini_async.py index 6c70efd9b..44e971e25 100644 --- a/posthog/ai/gemini/gemini_async.py +++ b/posthog/ai/gemini/gemini_async.py @@ -244,9 +244,27 @@ async def _generate_content_streaming( 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 diff --git a/posthog/ai/openai/openai.py b/posthog/ai/openai/openai.py index ccdf6e69a..07358c0bc 100644 --- a/posthog/ai/openai/openai.py +++ b/posthog/ai/openai/openai.py @@ -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 @@ -392,7 +406,21 @@ def _create_streaming( if "stream_options" not in kwargs: kwargs["stream_options"] = {} kwargs["stream_options"]["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 diff --git a/posthog/ai/openai/openai_async.py b/posthog/ai/openai/openai_async.py index 07e0de670..e3ff433c7 100644 --- a/posthog/ai/openai/openai_async.py +++ b/posthog/ai/openai/openai_async.py @@ -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 @@ -396,7 +410,21 @@ async def _create_streaming( if "stream_options" not in kwargs: kwargs["stream_options"] = {} kwargs["stream_options"]["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 diff --git a/posthog/test/ai/test_stream_midstream_error.py b/posthog/test/ai/test_stream_midstream_error.py index 1fbc8278f..7958b307d 100644 --- a/posthog/test/ai/test_stream_midstream_error.py +++ b/posthog/test/ai/test_stream_midstream_error.py @@ -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 From 29325a412843f29c24a975d9ea265c25ed8268cd Mon Sep 17 00:00:00 2001 From: Alex Castillo Date: Thu, 29 Oct 2026 21:13:23 -0400 Subject: [PATCH 2/4] fix(ai): annotate the Gemini streaming accumulator for mypy --- posthog/ai/gemini/gemini.py | 2 +- posthog/ai/gemini/gemini_async.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/posthog/ai/gemini/gemini.py b/posthog/ai/gemini/gemini.py index cacf8e4fc..4d06c554c 100644 --- a/posthog/ai/gemini/gemini.py +++ b/posthog/ai/gemini/gemini.py @@ -256,7 +256,7 @@ 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} diff --git a/posthog/ai/gemini/gemini_async.py b/posthog/ai/gemini/gemini_async.py index 44e971e25..bef6e3fe3 100644 --- a/posthog/ai/gemini/gemini_async.py +++ b/posthog/ai/gemini/gemini_async.py @@ -240,7 +240,7 @@ 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} From 3d813068cd30dd4a924a356728c8a49e1b88b892 Mon Sep 17 00:00:00 2001 From: Alex Castillo Date: Thu, 5 Nov 2026 10:50:25 -0400 Subject: [PATCH 3/4] fix(ai): keep the provider error when capturing a failed streaming connect If building the telemetry event raises (for example while formatting input the provider just rejected), the capture error replaced the original exception. Log it instead and re-raise the provider error. Signed-off-by: Alex Castillo --- posthog/ai/anthropic/anthropic.py | 33 +++++++++++++++---------- posthog/ai/anthropic/anthropic_async.py | 33 +++++++++++++++---------- 2 files changed, 40 insertions(+), 26 deletions(-) diff --git a/posthog/ai/anthropic/anthropic.py b/posthog/ai/anthropic/anthropic.py index 0267eb84f..30759fb04 100644 --- a/posthog/ai/anthropic/anthropic.py +++ b/posthog/ai/anthropic/anthropic.py @@ -6,6 +6,7 @@ "Please install the Anthropic SDK to use this feature: 'pip install anthropic'" ) +import logging import time import uuid from typing import Any, Dict, List, Optional @@ -31,6 +32,8 @@ from posthog.client import Client as PostHogClient from posthog import setup +log = logging.getLogger("posthog") + class Anthropic(anthropic.Anthropic): """ @@ -166,19 +169,23 @@ def _create_streaming( response = super().create(**kwargs) except Exception as exc: accumulator = _AnthropicStreamAccumulator() - 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, - ) + try: + 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, + ) + except Exception: + # Telemetry must never replace the provider's own error. + log.exception("Failed to capture streaming connect error") raise return self._track_streaming_response( response, diff --git a/posthog/ai/anthropic/anthropic_async.py b/posthog/ai/anthropic/anthropic_async.py index ce08ad0f2..dfe895356 100644 --- a/posthog/ai/anthropic/anthropic_async.py +++ b/posthog/ai/anthropic/anthropic_async.py @@ -6,11 +6,14 @@ "Please install the Anthropic SDK to use this feature: 'pip install anthropic'" ) +import logging import time import uuid from typing import Any, Dict, List, Optional from posthog import setup + +log = logging.getLogger("posthog") from ..stream import AsyncStreamWrapper as AsyncStreamWrapper from ..types import ( StreamingContentBlock as StreamingContentBlock, @@ -166,19 +169,23 @@ async def _create_streaming( 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, - ) + try: + 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, + ) + except Exception: + # Telemetry must never replace the provider's own error. + log.exception("Failed to capture streaming connect error") raise return self._track_streaming_response( response, From 8c738e11e197754142431a7f50e0462499cd35a7 Mon Sep 17 00:00:00 2001 From: Alex Castillo Date: Thu, 5 Nov 2026 18:57:23 -0400 Subject: [PATCH 4/4] fix(ai): do not let telemetry failures mask a streaming connect error If capturing the connect-error event itself raised (e.g. unformattable input), the capture exception replaced the provider error. Wrap the capture in each OpenAI/Gemini wrapper, log a fixed message, and re-raise the original error. The Anthropic fallback log no longer includes the exception text or traceback. Signed-off-by: Alex Castillo --- posthog/ai/anthropic/anthropic.py | 4 +- posthog/ai/anthropic/anthropic_async.py | 5 +- posthog/ai/gemini/gemini.py | 38 ++++++++----- posthog/ai/gemini/gemini_async.py | 38 ++++++++----- posthog/ai/openai/openai.py | 57 ++++++++++++------- posthog/ai/openai/openai_async.py | 57 ++++++++++++------- .../test/ai/test_stream_midstream_error.py | 50 ++++++++++++++++ 7 files changed, 171 insertions(+), 78 deletions(-) diff --git a/posthog/ai/anthropic/anthropic.py b/posthog/ai/anthropic/anthropic.py index 30759fb04..ef4c4dacb 100644 --- a/posthog/ai/anthropic/anthropic.py +++ b/posthog/ai/anthropic/anthropic.py @@ -32,7 +32,7 @@ from posthog.client import Client as PostHogClient from posthog import setup -log = logging.getLogger("posthog") +_log = logging.getLogger("posthog") class Anthropic(anthropic.Anthropic): @@ -185,7 +185,7 @@ def _create_streaming( ) except Exception: # Telemetry must never replace the provider's own error. - log.exception("Failed to capture streaming connect error") + _log.warning("[PostHog] Failed to capture streaming connect error") raise return self._track_streaming_response( response, diff --git a/posthog/ai/anthropic/anthropic_async.py b/posthog/ai/anthropic/anthropic_async.py index dfe895356..80014ce91 100644 --- a/posthog/ai/anthropic/anthropic_async.py +++ b/posthog/ai/anthropic/anthropic_async.py @@ -13,7 +13,6 @@ from posthog import setup -log = logging.getLogger("posthog") from ..stream import AsyncStreamWrapper as AsyncStreamWrapper from ..types import ( StreamingContentBlock as StreamingContentBlock, @@ -34,6 +33,8 @@ ) from posthog.client import Client as PostHogClient +_log = logging.getLogger("posthog") + class AsyncAnthropic(anthropic.AsyncAnthropic): """ @@ -185,7 +186,7 @@ async def _create_streaming( ) except Exception: # Telemetry must never replace the provider's own error. - log.exception("Failed to capture streaming connect error") + _log.warning("[PostHog] Failed to capture streaming connect error") raise return self._track_streaming_response( response, diff --git a/posthog/ai/gemini/gemini.py b/posthog/ai/gemini/gemini.py index b5ccb16aa..42d5cb688 100644 --- a/posthog/ai/gemini/gemini.py +++ b/posthog/ai/gemini/gemini.py @@ -1,3 +1,4 @@ +import logging import time from typing import Any, Dict, Optional @@ -29,6 +30,8 @@ format_gemini_streaming_output as format_gemini_streaming_output, ) +_log = logging.getLogger("posthog") + class Client: """ @@ -266,21 +269,26 @@ def _generate_content_streaming( **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, - ) + try: + 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, + ) + except Exception: + # Telemetry must never replace the provider's own error, and the + # fixed text avoids echoing prompt or error content into logs. + _log.warning("[PostHog] Failed to capture streaming connect error") raise def generator(): diff --git a/posthog/ai/gemini/gemini_async.py b/posthog/ai/gemini/gemini_async.py index 0ac476728..3b0d45c69 100644 --- a/posthog/ai/gemini/gemini_async.py +++ b/posthog/ai/gemini/gemini_async.py @@ -1,3 +1,4 @@ +import logging import time from typing import Any, Dict, Optional @@ -29,6 +30,8 @@ format_gemini_streaming_output as format_gemini_streaming_output, ) +_log = logging.getLogger("posthog") + class AsyncClient: """ @@ -250,21 +253,26 @@ async def _generate_content_streaming( **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, - ) + try: + 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, + ) + except Exception: + # Telemetry must never replace the provider's own error, and the + # fixed text avoids echoing prompt or error content into logs. + _log.warning("[PostHog] Failed to capture streaming connect error") raise async def async_generator(): diff --git a/posthog/ai/openai/openai.py b/posthog/ai/openai/openai.py index 2f58d1283..e00a2d3c0 100644 --- a/posthog/ai/openai/openai.py +++ b/posthog/ai/openai/openai.py @@ -1,3 +1,4 @@ +import logging import time import uuid from typing import TYPE_CHECKING as _TYPE_CHECKING, Any, Dict, Optional @@ -41,6 +42,8 @@ merge_provider_override, ) +_log = logging.getLogger("posthog") + class OpenAI(openai.OpenAI): """ @@ -170,17 +173,22 @@ def _create_streaming( 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, - ) + try: + 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, + ) + except Exception: + # Telemetry must never replace the provider's own error, and the + # fixed text avoids echoing prompt or error content into logs. + _log.warning("[PostHog] Failed to capture streaming connect error") raise def generator(): @@ -412,17 +420,22 @@ def _create_streaming( 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, - ) + try: + 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, + ) + except Exception: + # Telemetry must never replace the provider's own error, and the + # fixed text avoids echoing prompt or error content into logs. + _log.warning("[PostHog] Failed to capture streaming connect error") raise def generator(): diff --git a/posthog/ai/openai/openai_async.py b/posthog/ai/openai/openai_async.py index 2cd5f3028..3d105f7a2 100644 --- a/posthog/ai/openai/openai_async.py +++ b/posthog/ai/openai/openai_async.py @@ -1,3 +1,4 @@ +import logging import time import uuid from typing import TYPE_CHECKING as _TYPE_CHECKING, Any, Dict, Optional @@ -42,6 +43,8 @@ merge_provider_override, ) +_log = logging.getLogger("posthog") + class AsyncOpenAI(openai.AsyncOpenAI): """ @@ -171,17 +174,22 @@ async def _create_streaming( 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, - ) + try: + 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, + ) + except Exception: + # Telemetry must never replace the provider's own error, and the + # fixed text avoids echoing prompt or error content into logs. + _log.warning("[PostHog] Failed to capture streaming connect error") raise async def async_generator(): @@ -416,17 +424,22 @@ async def _create_streaming( 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, - ) + try: + 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, + ) + except Exception: + # Telemetry must never replace the provider's own error, and the + # fixed text avoids echoing prompt or error content into logs. + _log.warning("[PostHog] Failed to capture streaming connect error") raise async def async_generator(): diff --git a/posthog/test/ai/test_stream_midstream_error.py b/posthog/test/ai/test_stream_midstream_error.py index 95e363646..620b5990c 100644 --- a/posthog/test/ai/test_stream_midstream_error.py +++ b/posthog/test/ai/test_stream_midstream_error.py @@ -480,3 +480,53 @@ async def create(**kwargs): max_tokens=5, stream=True, ) + + +def test_openai_chat_stream_connect_error_survives_unformattable_input(mock_client): + from posthog.ai.openai import OpenAI + + with patch("openai.resources.chat.completions.Completions.create") as create: + create.side_effect = _ConnectError("bad request") + client = OpenAI(api_key="k", posthog_client=mock_client) + with pytest.raises(_ConnectError): + client.chat.completions.create(model="gpt-4", messages=123, stream=True) + + +@pytest.mark.asyncio +async def test_openai_chat_stream_connect_error_survives_unformattable_input_async( + mock_client, +): + from posthog.ai.openai import AsyncOpenAI + + async def create(**kwargs): + raise _ConnectError("bad request") + + 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=123, stream=True + ) + + +def test_anthropic_connect_error_fallback_does_not_log_error_text(mock_client, caplog): + from posthog.ai.anthropic import Anthropic + + with patch("anthropic.resources.messages.Messages.create") as create: + create.side_effect = _ConnectError("rejected PRIVATE_CANARY") + client = Anthropic(api_key="k", posthog_client=mock_client) + with caplog.at_level("DEBUG", logger="posthog"): + with pytest.raises(_ConnectError): + client.messages.create( + model="claude-3-opus-20240229", + messages=[None], + max_tokens=5, + stream=True, + posthog_privacy_mode=True, + ) + + assert "Failed to capture streaming connect error" in caplog.text + assert "PRIVATE_CANARY" not in caplog.text + assert all(r.exc_info is None for r in caplog.records)