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..ef4c4dacb 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): """ @@ -162,7 +165,28 @@ def _create_streaming( **kwargs: Any, ): start_time = time.time() - response = super().create(**kwargs) + try: + response = super().create(**kwargs) + except Exception as exc: + accumulator = _AnthropicStreamAccumulator() + 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.warning("[PostHog] Failed to capture streaming connect error") + 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..80014ce91 100644 --- a/posthog/ai/anthropic/anthropic_async.py +++ b/posthog/ai/anthropic/anthropic_async.py @@ -6,11 +6,13 @@ "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 + from ..stream import AsyncStreamWrapper as AsyncStreamWrapper from ..types import ( StreamingContentBlock as StreamingContentBlock, @@ -31,6 +33,8 @@ ) from posthog.client import Client as PostHogClient +_log = logging.getLogger("posthog") + class AsyncAnthropic(anthropic.AsyncAnthropic): """ @@ -162,7 +166,28 @@ 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() + 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.warning("[PostHog] Failed to capture streaming connect error") + 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 485e2adee..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: """ @@ -257,11 +260,36 @@ 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: + 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(): nonlocal usage_stats diff --git a/posthog/ai/gemini/gemini_async.py b/posthog/ai/gemini/gemini_async.py index 6823743ed..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: """ @@ -241,13 +244,36 @@ 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: + 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(): nonlocal usage_stats diff --git a/posthog/ai/openai/openai.py b/posthog/ai/openai/openai.py index 998e3fd15..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): """ @@ -167,7 +170,26 @@ def _create_streaming( ): start_time = time.time() state = _ResponsesStreamState() - response = self._original.create(**kwargs) + try: + response = self._original.create(**kwargs) + except Exception as 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(): stream_error: Optional[BaseException] = None @@ -395,7 +417,26 @@ 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: + 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(): stream_error: Optional[BaseException] = None diff --git a/posthog/ai/openai/openai_async.py b/posthog/ai/openai/openai_async.py index a37aebb6a..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): """ @@ -168,7 +171,26 @@ 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: + 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(): stream_error: Optional[BaseException] = None @@ -399,7 +421,26 @@ 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: + 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(): 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 386573a85..620b5990c 100644 --- a/posthog/test/ai/test_stream_midstream_error.py +++ b/posthog/test/ai/test_stream_midstream_error.py @@ -269,6 +269,168 @@ async def stream(): 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 + + @pytest.mark.parametrize( ("cls_name", "code"), [("ClientError", 429), ("ServerError", 503)] ) @@ -283,3 +445,88 @@ def test_capture_streaming_event_records_google_genai_error_status( props = _props(mock_client) assert props["$ai_http_status"] == code assert props["$ai_is_error"] is True + + +def test_anthropic_stream_connect_error_survives_unformattable_input(mock_client): + from posthog.ai.anthropic import Anthropic + + with patch("anthropic.resources.messages.Messages.create") as create: + create.side_effect = _ConnectError("bad request") + client = Anthropic(api_key="k", posthog_client=mock_client) + with pytest.raises(_ConnectError): + client.messages.create( + model="claude-3-opus-20240229", + messages=[None], + max_tokens=5, + stream=True, + ) + + +@pytest.mark.asyncio +async def test_anthropic_stream_connect_error_survives_unformattable_input_async( + mock_client, +): + from posthog.ai.anthropic import AsyncAnthropic + + async def create(**kwargs): + raise _ConnectError("bad request") + + 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=[None], + 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)