diff --git a/sentry_sdk/integrations/openai.py b/sentry_sdk/integrations/openai.py index 742b693245..31f2bb873b 100644 --- a/sentry_sdk/integrations/openai.py +++ b/sentry_sdk/integrations/openai.py @@ -148,6 +148,11 @@ def setup_once() -> None: Responses.create = _wrap_responses_create(Responses.create) # type: ignore[method-assign] AsyncResponses.create = _wrap_async_responses_create(AsyncResponses.create) # type: ignore[assignment,method-assign] + # Finishing the span on close() covers streams that are closed early, + # abandoned mid-iteration, or exited via `with` before exhaustion. + Stream.close = _wrap_close(Stream.close) # type: ignore[assignment,method-assign] + AsyncStream.close = _wrap_async_close(AsyncStream.close) # type: ignore[assignment,method-assign] + def count_tokens(self: "OpenAIIntegration", s: str) -> int: if self.tiktoken_encoding is None: return 0 @@ -730,7 +735,7 @@ def _set_common_output_data( if record_outputs: for output in response.output: if output.type == "function_call": - output_messages["tool"].append(output.dict()) + output_messages["tool"].append(output.model_dump()) elif output.type == "message": for output_message in output.content: try: @@ -738,7 +743,7 @@ def _set_common_output_data( except AttributeError: # Unknown output message type, just return the json output_messages["response"].append( - output_message.dict() + output_message.model_dump() ) if len(output_messages["tool"]) > 0: @@ -761,14 +766,16 @@ def _set_common_output_data( ): for output in response.output: if output.type == "function_call": - output_messages["tool"].append(output.dict()) + output_messages["tool"].append(output.model_dump()) elif output.type == "message": for output_message in output.content: try: output_messages["response"].append(output_message.text) # type: ignore[union-attr] except AttributeError: # Unknown output message type, just return the json - output_messages["response"].append(output_message.dict()) + output_messages["response"].append( + output_message.model_dump() + ) if len(output_messages["tool"]) > 0: set_data_normalized( @@ -885,6 +892,8 @@ def _new_sync_chat_completion( old_iterator=response._iterator, finish_span=True, ) + # Attach the span so the patched Stream.close() can finish it on early exit. + response._span = span # type: ignore[attr-defined] else: _set_completions_api_output_data( @@ -969,6 +978,8 @@ async def _new_async_chat_completion( old_iterator=response._iterator, finish_span=True, ) + # Attach the span so the patched Stream.close() can finish it on early exit. + response._span = span # type: ignore[attr-defined] else: _set_completions_api_output_data( span, response, kwargs, integration, finish_span=True @@ -998,6 +1009,59 @@ def _set_completions_api_output_data( ) +def _finish_stream_span( + stream: "Union[Stream[Any], AsyncStream[Any]]", + exc_info: "tuple[Any, Any, Any]", +) -> None: + """ + Finishes the AI Client Span attached to a stream, if any, and detaches it + so the span is finished exactly once (close() and iterator finalization + both route through here). + """ + span = getattr(stream, "_span", None) + if span is None: + return + with capture_internal_exceptions(): + del stream._span # type: ignore[union-attr] + span.__exit__(*exc_info) + + +def _wrap_close( + f: "Callable[[Stream[Any]], None]", +) -> "Callable[[Stream[Any]], None]": + """ + Finishes the stream's AI Client Span when the stream is closed early, + unless the chunk iterator already finished it. + """ + + @wraps(f) + def close(self: "Stream[Any]") -> None: + try: + return f(self) + finally: + _finish_stream_span(self, sys.exc_info()) + + return close + + +def _wrap_async_close( + f: "Callable[[AsyncStream[Any]], Coroutine[Any, Any, None]]", +) -> "Callable[[AsyncStream[Any]], Coroutine[Any, Any, None]]": + """ + Finishes the stream's AI Client Span when the stream is closed early, + unless the chunk iterator already finished it. + """ + + @wraps(f) + async def close(self: "AsyncStream[Any]") -> None: + try: + return await f(self) + finally: + _finish_stream_span(self, sys.exc_info()) + + return close + + def _wrap_synchronous_completions_chunk_iterator( span: "Union[Span, StreamedSpan]", integration: "OpenAIIntegration", @@ -1017,56 +1081,64 @@ def _wrap_synchronous_completions_chunk_iterator( streaming_message_total_token_usage = None client = sentry_sdk.get_client() - for x in old_iterator: - if isinstance(span, StreamedSpan): - span.set_attribute(SPANDATA.GEN_AI_RESPONSE_MODEL, x.model) - else: - span.set_data(SPANDATA.GEN_AI_RESPONSE_MODEL, x.model) + try: + for x in old_iterator: + if isinstance(span, StreamedSpan): + span.set_attribute(SPANDATA.GEN_AI_RESPONSE_MODEL, x.model) + else: + span.set_data(SPANDATA.GEN_AI_RESPONSE_MODEL, x.model) + + with capture_internal_exceptions(): + if hasattr(x, "choices") and x.choices is not None: + choice_index = 0 + for choice in x.choices: + if hasattr(choice, "delta") and hasattr( + choice.delta, "content" + ): + if start_time is not None and ttft is None: + ttft = time.perf_counter() - start_time + content = choice.delta.content + if len(data_buf) <= choice_index: + data_buf.append([]) + data_buf[choice_index].append(content or "") + choice_index += 1 + if hasattr(x, "usage"): + streaming_message_total_token_usage = x.usage + + yield x with capture_internal_exceptions(): - if hasattr(x, "choices") and x.choices is not None: - choice_index = 0 - for choice in x.choices: - if hasattr(choice, "delta") and hasattr(choice.delta, "content"): - if start_time is not None and ttft is None: - ttft = time.perf_counter() - start_time - content = choice.delta.content - if len(data_buf) <= choice_index: - data_buf.append([]) - data_buf[choice_index].append(content or "") - choice_index += 1 - if hasattr(x, "usage"): - streaming_message_total_token_usage = x.usage - - yield x - - with capture_internal_exceptions(): - if ttft is not None: - set_data_normalized( - span, SPANDATA.GEN_AI_RESPONSE_TIME_TO_FIRST_TOKEN, ttft - ) - all_responses = None - if len(data_buf) > 0: - all_responses = ["".join(chunk) for chunk in data_buf] - if has_data_collection_enabled(client.options): - if client.options["data_collection"]["gen_ai"]["outputs"]: + if ttft is not None: + set_data_normalized( + span, SPANDATA.GEN_AI_RESPONSE_TIME_TO_FIRST_TOKEN, ttft + ) + all_responses = None + if len(data_buf) > 0: + all_responses = ["".join(chunk) for chunk in data_buf] + if has_data_collection_enabled(client.options): + if client.options["data_collection"]["gen_ai"]["outputs"]: + set_data_normalized( + span, SPANDATA.GEN_AI_RESPONSE_TEXT, all_responses + ) + elif should_send_default_pii() and integration.include_prompts: set_data_normalized( span, SPANDATA.GEN_AI_RESPONSE_TEXT, all_responses ) - elif should_send_default_pii() and integration.include_prompts: - set_data_normalized(span, SPANDATA.GEN_AI_RESPONSE_TEXT, all_responses) - _calculate_completions_token_usage( - messages=messages, - response=response, - span=span, - streaming_message_responses=all_responses, - streaming_message_total_token_usage=streaming_message_total_token_usage, - count_tokens=integration.count_tokens, - ) - - if finish_span: - span.__exit__(None, None, None) + _calculate_completions_token_usage( + messages=messages, + response=response, + span=span, + streaming_message_responses=all_responses, + streaming_message_total_token_usage=streaming_message_total_token_usage, + count_tokens=integration.count_tokens, + ) + finally: + # Finish the span even if the stream is closed early, abandoned, or + # the connection drops mid-stream. close() detaches the span first, + # making this a no-op if it already ran there. + if finish_span: + _finish_stream_span(response, sys.exc_info()) async def _wrap_asynchronous_completions_chunk_iterator( @@ -1088,56 +1160,64 @@ async def _wrap_asynchronous_completions_chunk_iterator( streaming_message_total_token_usage = None client = sentry_sdk.get_client() - async for x in old_iterator: - if isinstance(span, StreamedSpan): - span.set_attribute(SPANDATA.GEN_AI_RESPONSE_MODEL, x.model) - else: - span.set_data(SPANDATA.GEN_AI_RESPONSE_MODEL, x.model) + try: + async for x in old_iterator: + if isinstance(span, StreamedSpan): + span.set_attribute(SPANDATA.GEN_AI_RESPONSE_MODEL, x.model) + else: + span.set_data(SPANDATA.GEN_AI_RESPONSE_MODEL, x.model) + + with capture_internal_exceptions(): + if hasattr(x, "choices") and x.choices is not None: + choice_index = 0 + for choice in x.choices: + if hasattr(choice, "delta") and hasattr( + choice.delta, "content" + ): + if start_time is not None and ttft is None: + ttft = time.perf_counter() - start_time + content = choice.delta.content + if len(data_buf) <= choice_index: + data_buf.append([]) + data_buf[choice_index].append(content or "") + choice_index += 1 + if hasattr(x, "usage"): + streaming_message_total_token_usage = x.usage + + yield x with capture_internal_exceptions(): - if hasattr(x, "choices") and x.choices is not None: - choice_index = 0 - for choice in x.choices: - if hasattr(choice, "delta") and hasattr(choice.delta, "content"): - if start_time is not None and ttft is None: - ttft = time.perf_counter() - start_time - content = choice.delta.content - if len(data_buf) <= choice_index: - data_buf.append([]) - data_buf[choice_index].append(content or "") - choice_index += 1 - if hasattr(x, "usage"): - streaming_message_total_token_usage = x.usage - - yield x - - with capture_internal_exceptions(): - if ttft is not None: - set_data_normalized( - span, SPANDATA.GEN_AI_RESPONSE_TIME_TO_FIRST_TOKEN, ttft - ) - all_responses = None - if len(data_buf) > 0: - all_responses = ["".join(chunk) for chunk in data_buf] - if has_data_collection_enabled(client.options): - if client.options["data_collection"]["gen_ai"]["outputs"]: + if ttft is not None: + set_data_normalized( + span, SPANDATA.GEN_AI_RESPONSE_TIME_TO_FIRST_TOKEN, ttft + ) + all_responses = None + if len(data_buf) > 0: + all_responses = ["".join(chunk) for chunk in data_buf] + if has_data_collection_enabled(client.options): + if client.options["data_collection"]["gen_ai"]["outputs"]: + set_data_normalized( + span, SPANDATA.GEN_AI_RESPONSE_TEXT, all_responses + ) + elif should_send_default_pii() and integration.include_prompts: set_data_normalized( span, SPANDATA.GEN_AI_RESPONSE_TEXT, all_responses ) - elif should_send_default_pii() and integration.include_prompts: - set_data_normalized(span, SPANDATA.GEN_AI_RESPONSE_TEXT, all_responses) - _calculate_completions_token_usage( - messages=messages, - response=response, - span=span, - streaming_message_responses=all_responses, - streaming_message_total_token_usage=streaming_message_total_token_usage, - count_tokens=integration.count_tokens, - ) - - if finish_span: - span.__exit__(None, None, None) + _calculate_completions_token_usage( + messages=messages, + response=response, + span=span, + streaming_message_responses=all_responses, + streaming_message_total_token_usage=streaming_message_total_token_usage, + count_tokens=integration.count_tokens, + ) + finally: + # Finish the span even if the stream is closed early, abandoned, or + # the connection drops mid-stream. close() detaches the span first, + # making this a no-op if it already ran there. + if finish_span: + _finish_stream_span(response, sys.exc_info()) def _wrap_synchronous_responses_event_iterator( @@ -1159,58 +1239,66 @@ def _wrap_synchronous_responses_event_iterator( client = sentry_sdk.get_client() count_tokens_manually = True - for x in old_iterator: - with capture_internal_exceptions(): - if isinstance(x, ResponseTextDeltaEvent): - if start_time is not None and ttft is None: - ttft = time.perf_counter() - start_time - if len(data_buf) == 0: - data_buf.append([]) - data_buf[0].append(x.delta or "") - - elif isinstance(x, ResponseCompletedEvent): - if isinstance(span, StreamedSpan): - span.set_attribute(SPANDATA.GEN_AI_RESPONSE_MODEL, x.response.model) - else: - span.set_data(SPANDATA.GEN_AI_RESPONSE_MODEL, x.response.model) - - _calculate_responses_token_usage( - input=input, - response=x.response, - span=span, - streaming_message_responses=None, - count_tokens=integration.count_tokens, - ) - count_tokens_manually = False + try: + for x in old_iterator: + with capture_internal_exceptions(): + if isinstance(x, ResponseTextDeltaEvent): + if start_time is not None and ttft is None: + ttft = time.perf_counter() - start_time + if len(data_buf) == 0: + data_buf.append([]) + data_buf[0].append(x.delta or "") + + elif isinstance(x, ResponseCompletedEvent): + if isinstance(span, StreamedSpan): + span.set_attribute( + SPANDATA.GEN_AI_RESPONSE_MODEL, x.response.model + ) + else: + span.set_data(SPANDATA.GEN_AI_RESPONSE_MODEL, x.response.model) + + _calculate_responses_token_usage( + input=input, + response=x.response, + span=span, + streaming_message_responses=None, + count_tokens=integration.count_tokens, + ) + count_tokens_manually = False - yield x + yield x - with capture_internal_exceptions(): - if ttft is not None: - set_data_normalized( - span, SPANDATA.GEN_AI_RESPONSE_TIME_TO_FIRST_TOKEN, ttft - ) - if len(data_buf) > 0: - all_responses = ["".join(chunk) for chunk in data_buf] - if has_data_collection_enabled(client.options): - if client.options["data_collection"]["gen_ai"]["outputs"]: + with capture_internal_exceptions(): + if ttft is not None: + set_data_normalized( + span, SPANDATA.GEN_AI_RESPONSE_TIME_TO_FIRST_TOKEN, ttft + ) + if len(data_buf) > 0: + all_responses = ["".join(chunk) for chunk in data_buf] + if has_data_collection_enabled(client.options): + if client.options["data_collection"]["gen_ai"]["outputs"]: + set_data_normalized( + span, SPANDATA.GEN_AI_RESPONSE_TEXT, all_responses + ) + elif should_send_default_pii() and integration.include_prompts: set_data_normalized( span, SPANDATA.GEN_AI_RESPONSE_TEXT, all_responses ) - elif should_send_default_pii() and integration.include_prompts: - set_data_normalized(span, SPANDATA.GEN_AI_RESPONSE_TEXT, all_responses) - - if count_tokens_manually: - _calculate_responses_token_usage( - input=input, - response=response, - span=span, - streaming_message_responses=all_responses, - count_tokens=integration.count_tokens, - ) - if finish_span: - span.__exit__(None, None, None) + if count_tokens_manually: + _calculate_responses_token_usage( + input=input, + response=response, + span=span, + streaming_message_responses=all_responses, + count_tokens=integration.count_tokens, + ) + finally: + # Finish the span even if the stream is closed early, abandoned, or + # the connection drops mid-stream. close() detaches the span first, + # making this a no-op if it already ran there. + if finish_span: + _finish_stream_span(response, sys.exc_info()) async def _wrap_asynchronous_responses_event_iterator( @@ -1232,58 +1320,67 @@ async def _wrap_asynchronous_responses_event_iterator( client = sentry_sdk.get_client() count_tokens_manually = True - async for x in old_iterator: - with capture_internal_exceptions(): - if isinstance(x, ResponseTextDeltaEvent): - if start_time is not None and ttft is None: - ttft = time.perf_counter() - start_time - if len(data_buf) == 0: - data_buf.append([]) - data_buf[0].append(x.delta or "") - - elif isinstance(x, ResponseCompletedEvent): - if isinstance(span, StreamedSpan): - span.set_attribute(SPANDATA.GEN_AI_RESPONSE_MODEL, x.response.model) - else: - span.set_data(SPANDATA.GEN_AI_RESPONSE_MODEL, x.response.model) - - _calculate_responses_token_usage( - input=input, - response=x.response, - span=span, - streaming_message_responses=None, - count_tokens=integration.count_tokens, - ) - count_tokens_manually = False + try: + async for x in old_iterator: + with capture_internal_exceptions(): + if isinstance(x, ResponseTextDeltaEvent): + if start_time is not None and ttft is None: + ttft = time.perf_counter() - start_time + if len(data_buf) == 0: + data_buf.append([]) + data_buf[0].append(x.delta or "") + + elif isinstance(x, ResponseCompletedEvent): + if isinstance(span, StreamedSpan): + span.set_attribute( + SPANDATA.GEN_AI_RESPONSE_MODEL, x.response.model + ) + else: + span.set_data(SPANDATA.GEN_AI_RESPONSE_MODEL, x.response.model) + + _calculate_responses_token_usage( + input=input, + response=x.response, + span=span, + streaming_message_responses=None, + count_tokens=integration.count_tokens, + ) + count_tokens_manually = False - yield x + yield x - with capture_internal_exceptions(): - if ttft is not None: - set_data_normalized( - span, SPANDATA.GEN_AI_RESPONSE_TIME_TO_FIRST_TOKEN, ttft - ) - if len(data_buf) > 0: - all_responses = ["".join(chunk) for chunk in data_buf] - - if has_data_collection_enabled(client.options): - if client.options["data_collection"]["gen_ai"]["outputs"]: + with capture_internal_exceptions(): + if ttft is not None: + set_data_normalized( + span, SPANDATA.GEN_AI_RESPONSE_TIME_TO_FIRST_TOKEN, ttft + ) + if len(data_buf) > 0: + all_responses = ["".join(chunk) for chunk in data_buf] + + if has_data_collection_enabled(client.options): + if client.options["data_collection"]["gen_ai"]["outputs"]: + set_data_normalized( + span, SPANDATA.GEN_AI_RESPONSE_TEXT, all_responses + ) + elif should_send_default_pii() and integration.include_prompts: set_data_normalized( span, SPANDATA.GEN_AI_RESPONSE_TEXT, all_responses ) - elif should_send_default_pii() and integration.include_prompts: - set_data_normalized(span, SPANDATA.GEN_AI_RESPONSE_TEXT, all_responses) - - if count_tokens_manually: - _calculate_responses_token_usage( - input=input, - response=response, - span=span, - streaming_message_responses=all_responses, - count_tokens=integration.count_tokens, - ) - if finish_span: - span.__exit__(None, None, None) + + if count_tokens_manually: + _calculate_responses_token_usage( + input=input, + response=response, + span=span, + streaming_message_responses=all_responses, + count_tokens=integration.count_tokens, + ) + finally: + # Finish the span even if the stream is closed early, abandoned, or + # the connection drops mid-stream. close() detaches the span first, + # making this a no-op if it already ran there. + if finish_span: + _finish_stream_span(response, sys.exc_info()) def _set_responses_api_output_data( @@ -1567,6 +1664,8 @@ def _new_sync_responses_create( old_iterator=response._iterator, finish_span=True, ) + # Attach the span so the patched Stream.close() can finish it on early exit. + response._span = span # type: ignore[attr-defined] else: _set_responses_api_output_data( @@ -1643,6 +1742,8 @@ async def _new_async_responses_create( old_iterator=response._iterator, finish_span=True, ) + # Attach the span so the patched Stream.close() can finish it on early exit. + response._span = span # type: ignore[attr-defined] else: _set_responses_api_output_data( span, response, kwargs, integration, finish_span=True diff --git a/sentry_sdk/integrations/openai_agents/utils.py b/sentry_sdk/integrations/openai_agents/utils.py index 01303e0e58..0e160193ed 100644 --- a/sentry_sdk/integrations/openai_agents/utils.py +++ b/sentry_sdk/integrations/openai_agents/utils.py @@ -226,14 +226,14 @@ def _set_output_data( for output in result.output: if output.type == "function_call": - output_messages["tool"].append(output.dict()) + output_messages["tool"].append(output.model_dump()) elif output.type == "message": for output_message in output.content: try: output_messages["response"].append(output_message.text) except AttributeError: # Unknown output message type, just return the json - output_messages["response"].append(output_message.dict()) + output_messages["response"].append(output_message.model_dump()) if record_outputs and len(output_messages["tool"]) > 0: if isinstance(span, StreamedSpan): diff --git a/sentry_sdk/scope.py b/sentry_sdk/scope.py index 0d3a0c6cd3..e3a8460a21 100644 --- a/sentry_sdk/scope.py +++ b/sentry_sdk/scope.py @@ -912,6 +912,12 @@ def span(self, span: "Optional[Span]") -> None: self._transaction = transaction.name if transaction.source: self._transaction_info["source"] = transaction.source + elif span is None: + self._transaction = None + # The transaction source describes the transaction name; leaving + # it behind would stamp stale transaction_info on later events + # that no longer carry a transaction. + self._transaction_info = {} @property def streamed_span(self) -> "Optional[StreamedSpan]": diff --git a/sentry_sdk/transport.py b/sentry_sdk/transport.py index 3f89e8c418..abc8bd48a7 100644 --- a/sentry_sdk/transport.py +++ b/sentry_sdk/transport.py @@ -211,6 +211,15 @@ def _parse_rate_limits( continue +class TransportResponseError(Exception): + """Raised when the HTTP transport receives an unexpected response status. + + A non-2xx status (other than 429 rate limiting) means the envelope was not + accepted. Raising lets custom/wrapper transports detect the failed + delivery and fail over to another route instead of assuming success. + """ + + class HttpTransportCore(Transport): """Shared base class for sync and async transports.""" @@ -404,6 +413,13 @@ def _handle_response( self._handle_request_error( envelope=envelope, loss_reason="status_{}".format(response.status) ) + # Surface the failed delivery so custom/wrapper transports can + # catch it and fail over (e.g. a 403/404 from a network middlebox). + # 429 is handled via rate limits above and 413 is a deliberate + # drop, so neither raises here. + raise TransportResponseError( + "Unexpected status code: {}".format(response.status) + ) def _update_headers( self: "Self", diff --git a/tests/integrations/openai/test_openai.py b/tests/integrations/openai/test_openai.py index 96866886ec..80d1dc022a 100644 --- a/tests/integrations/openai/test_openai.py +++ b/tests/integrations/openai/test_openai.py @@ -2568,6 +2568,86 @@ def test_streaming_chat_completion( pass # if tiktoken is not installed, we can't guarantee token usage will be calculated properly +@pytest.mark.parametrize("consume", ["break_then_close", "with_block"]) +def test_streaming_chat_completion_span_finished_on_early_exit( + sentry_init, + capture_events, + get_model_response, + server_side_event_chunks, + consume, +): + # https://github.com/getsentry/sentry-python/issues/7847 + # Closing a stream early (or leaving a `with` block after one chunk) must + # still finish the gen_ai.chat span. + sentry_init( + integrations=[OpenAIIntegration()], + disabled_integrations=[StdlibIntegration], + traces_sample_rate=1.0, + # Keep gen_ai spans on the transaction so the assertion below sees them. + stream_gen_ai_spans=False, + ) + + client = OpenAI(api_key="z") + returned_stream = get_model_response( + server_side_event_chunks( + [ + ChatCompletionChunk( + id="1", + choices=[ + DeltaChoice( + index=0, + delta=ChoiceDelta(content="hel"), + finish_reason=None, + ) + ], + created=100000, + model="model-id", + object="chat.completion.chunk", + ), + ChatCompletionChunk( + id="1", + choices=[ + DeltaChoice( + index=0, + delta=ChoiceDelta(content="lo"), + finish_reason=None, + ) + ], + created=100000, + model="model-id", + object="chat.completion.chunk", + ), + ], + include_event_type=False, + ) + ) + + with mock.patch.object( + client.chat._client._client, + "send", + return_value=returned_stream, + ): + events = capture_events() + with start_transaction(name="openai tx"): + stream = client.chat.completions.create( + model="some-model", + messages=[{"role": "user", "content": "hello"}], + stream=True, + ) + if consume == "break_then_close": + for _ in stream: + break + stream.close() + else: + with stream: + next(iter(stream)) + + transactions = [event for event in events if event.get("type") == "transaction"] + assert len(transactions) == 1 + span_ops = [span["op"] for span in transactions[0]["spans"]] + assert "gen_ai.chat" in span_ops + + # noinspection PyTypeChecker @pytest.mark.parametrize("span_streaming", [True, False]) @pytest.mark.parametrize("stream_gen_ai_spans", [True, False]) diff --git a/tests/integrations/openai_agents/test_openai_agents.py b/tests/integrations/openai_agents/test_openai_agents.py index 8bc49a2c9f..13a336441d 100644 --- a/tests/integrations/openai_agents/test_openai_agents.py +++ b/tests/integrations/openai_agents/test_openai_agents.py @@ -2,6 +2,8 @@ import json import logging import os +import warnings +from types import SimpleNamespace from unittest import mock from unittest.mock import MagicMock, patch @@ -20,6 +22,7 @@ from agents.items import ( ResponseFunctionToolCall, ResponseOutputMessage, + ResponseOutputRefusal, ResponseOutputText, ) from agents.models.openai_responses import OpenAIResponsesModel @@ -72,7 +75,11 @@ from sentry_sdk.consts import OP, SPANDATA from sentry_sdk.integrations.logging import LoggingIntegration from sentry_sdk.integrations.openai_agents import OpenAIAgentsIntegration -from sentry_sdk.integrations.openai_agents.utils import _set_input_data, safe_serialize +from sentry_sdk.integrations.openai_agents.utils import ( + _set_input_data, + _set_output_data, + safe_serialize, +) from sentry_sdk.integrations.stdlib import StdlibIntegration from sentry_sdk.utils import package_version, parse_version @@ -7483,3 +7490,45 @@ async def test_runner_run_streamed_with_starting_agent_kwarg( (transaction,) = events assert transaction["transaction"] == "test_agent workflow" + + +def test_set_output_data_tool_calls_no_pydantic_deprecation(sentry_init): + """Serializing tool-call output must not use the deprecated pydantic v1 API. + + Regression test for GH-7827: _set_output_data called BaseModel.dict(), + which raises PydanticDeprecatedSince20 when DeprecationWarnings are errors. + """ + sentry_init( + integrations=[OpenAIAgentsIntegration()], + send_default_pii=True, + traces_sample_rate=1.0, + ) + + tool_call = ResponseFunctionToolCall( + type="function_call", + id="call_1", + call_id="call_1", + name="get_weather", + arguments='{"city": "Berlin"}', + status="completed", + ) + refusal = ResponseOutputRefusal(type="refusal", refusal="I cannot help.") + message = ResponseOutputMessage( + type="message", + id="msg_1", + role="assistant", + content=[refusal], + status="completed", + ) + result = SimpleNamespace(output=[tool_call, message]) + + with start_span(op="test") as span: + with warnings.catch_warnings(): + warnings.simplefilter("error", DeprecationWarning) + _set_output_data(span, result) + + tool_calls = json.loads(span._data[SPANDATA.GEN_AI_RESPONSE_TOOL_CALLS]) + assert tool_calls[0]["name"] == "get_weather" + assert tool_calls[0]["arguments"] == '{"city": "Berlin"}' + + assert "I cannot help." in span._data[SPANDATA.GEN_AI_RESPONSE_TEXT] diff --git a/tests/test_scope.py b/tests/test_scope.py index f05f17b313..78ef3a7b54 100644 --- a/tests/test_scope.py +++ b/tests/test_scope.py @@ -1110,3 +1110,45 @@ def test_conversation_id_clear(): scope.clear() assert scope.get_conversation_id() is None + + +def test_transaction_name_cleared_when_span_unset(): + """Scope._transaction follows the span lifecycle (GH-7774). + + Set by a Transaction, preserved while a child span is active, cleared + when the span is set back to None (top-level transaction exit). + """ + from sentry_sdk.tracing import Transaction + + scope = Scope() + + transaction = Transaction(name="my-transaction") + scope.span = transaction + assert scope._transaction == "my-transaction" + assert scope._transaction_info.get("source") # info follows the name + + child = transaction.start_child(op="child") + scope.span = child + assert scope._transaction == "my-transaction" + + scope.span = None + assert scope._transaction is None + assert scope._transaction_info == {} # no stale source left behind + + +def test_transaction_name_survives_nested_span_exit(): + """Exiting a child span restores the transaction on the scope. + + The transaction name must not be dropped when an inner span finishes + while the transaction is still live. + """ + scope = Scope() + + with sentry_sdk.start_transaction(name="my-transaction", scope=scope): + assert scope._transaction == "my-transaction" + with sentry_sdk.start_span(op="child", scope=scope): + assert scope._transaction == "my-transaction" + # span restored to the transaction, not None + assert scope._transaction == "my-transaction" + + assert scope._transaction is None diff --git a/tests/test_transport.py b/tests/test_transport.py index 3db95c06f6..f21a629727 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -1110,3 +1110,40 @@ async def send_message(i): # New request should be dropped due to rate limiting assert len(capturing_server.captured) == 0 await client.close_async(timeout=2.0) + + +from sentry_sdk.consts import EndpointType +from sentry_sdk.transport import TransportResponseError + + +@pytest.mark.parametrize("status_code", [403, 404, 500]) +def test_unexpected_status_code_raises(capturing_server, make_client, status_code): + # https://github.com/getsentry/sentry-python/issues/7844 + # A non-2xx response (e.g. 403/404 from a network middlebox) is a failed + # delivery and must raise, so wrapper transports can catch it and fail + # over instead of assuming the event was sent. + client = make_client() + capturing_server.respond_with(code=status_code) + + with pytest.raises(TransportResponseError, match="Unexpected status code"): + client.transport._send_request( + body=b"{}", + headers={}, + endpoint_type=EndpointType.ENVELOPE, + ) + + +@pytest.mark.parametrize("status_code", [413, 429]) +def test_expected_error_status_codes_do_not_raise( + capturing_server, make_client, status_code +): + # 413 (envelope too large) is a deliberate drop and 429 is handled via + # rate limits; neither should raise. + client = make_client() + capturing_server.respond_with(code=status_code) + + client.transport._send_request( + body=b"{}", + headers={}, + endpoint_type=EndpointType.ENVELOPE, + )