From 9879c76db8611a02d2faeabddea3abe19ba299e8 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 22 Sep 2026 09:15:01 +0200 Subject: [PATCH] ref(openai-agents): Remove workflow spans --- .../openai_agents/patches/agent_run.py | 34 +- .../openai_agents/patches/runner.py | 180 ++++------ .../openai_agents/spans/__init__.py | 1 - .../openai_agents/spans/agent_workflow.py | 16 - .../openai_agents/test_openai_agents.py | 313 +----------------- 5 files changed, 69 insertions(+), 475 deletions(-) delete mode 100644 sentry_sdk/integrations/openai_agents/spans/agent_workflow.py diff --git a/sentry_sdk/integrations/openai_agents/patches/agent_run.py b/sentry_sdk/integrations/openai_agents/patches/agent_run.py index 5a1c5deecc..8be5a97685 100644 --- a/sentry_sdk/integrations/openai_agents/patches/agent_run.py +++ b/sentry_sdk/integrations/openai_agents/patches/agent_run.py @@ -36,14 +36,6 @@ def _get_current_agent( return getattr(context_wrapper, "_sentry_current_agent", None) -def _close_streaming_workflow_span(agent: "Optional[agents.Agent]") -> None: - """Close the workflow span for streaming executions if it exists.""" - if agent and hasattr(agent, "_sentry_workflow_span"): - workflow_span = agent._sentry_workflow_span - workflow_span.__exit__(*sys.exc_info()) - delattr(agent, "_sentry_workflow_span") - - def _maybe_start_agent_span( context_wrapper: "Optional[agents.RunContextWrapper]", agent: "Optional[agents.Agent]", @@ -201,7 +193,6 @@ async def _run_single_turn_streamed( update_invoke_agent_span(span=span, agent=agent) span.__exit__(*exc_info) delattr(context_wrapper, "_sentry_agent_span") - _close_streaming_workflow_span(agent) reraise(*exc_info) return result @@ -216,7 +207,6 @@ async def _execute_handoffs( Patched execute_handoffs that - creates and manages handoff spans. - ends the agent invocation span. - - ends the workflow span if the response is streamed and an exception is raised in `execute_handoffs()`. """ context_wrapper: "Optional[agents.RunContextWrapper]" = kwargs.get( @@ -232,23 +222,12 @@ async def _execute_handoffs( handoff_agent_name = first_handoff.handoff.agent_name handoff_span(context_wrapper, agent, handoff_agent_name) - if not agent or not context_wrapper or not _has_active_agent_span(context_wrapper): - # Call original method with all parameters - try: - return await original_execute_handoffs(*args, **kwargs) - except Exception: - exc_info = sys.exc_info() - with capture_internal_exceptions(): - _close_streaming_workflow_span(agent) - reraise(*exc_info) - # Call original method with all parameters try: result = await original_execute_handoffs(*args, **kwargs) except Exception: exc_info = sys.exc_info() with capture_internal_exceptions(): - _close_streaming_workflow_span(agent) span = getattr(context_wrapper, "_sentry_agent_span", None) if span: update_invoke_agent_span(span=span, agent=agent) @@ -271,9 +250,7 @@ async def _execute_final_output( **kwargs: "Any", ) -> "SingleStepResult": """ - Patched execute_final_output that - - ends the agent invocation span. - - ends the workflow span if the response is streamed. + Patched execute_final_output that ends the agent invocation span. """ # openai-agents >= 0.14 renamed `agent` to `public_agent`. @@ -282,20 +259,13 @@ async def _execute_final_output( final_output = kwargs.get("final_output") if not agent or not context_wrapper or not _has_active_agent_span(context_wrapper): - try: - return await original_execute_final_output(*args, **kwargs) - finally: - with capture_internal_exceptions(): - # For streaming, close the workflow span (non-streaming uses context manager in _create_run_wrapper) - _close_streaming_workflow_span(agent) + return await original_execute_final_output(*args, **kwargs) try: result = await original_execute_final_output(*args, **kwargs) except Exception: exc_info = sys.exc_info() with capture_internal_exceptions(): - # For streaming, close the workflow span (non-streaming uses context manager in _create_run_wrapper) - _close_streaming_workflow_span(agent) span = getattr(context_wrapper, "_sentry_agent_span", None) if span: update_invoke_agent_span(span=span, agent=agent, output=final_output) diff --git a/sentry_sdk/integrations/openai_agents/patches/runner.py b/sentry_sdk/integrations/openai_agents/patches/runner.py index fe225cb9a0..66c3482695 100644 --- a/sentry_sdk/integrations/openai_agents/patches/runner.py +++ b/sentry_sdk/integrations/openai_agents/patches/runner.py @@ -12,7 +12,6 @@ ) from ..spans import ( - agent_workflow_span, execute_tool_span, update_execute_tool_span, update_invoke_agent_span, @@ -28,7 +27,7 @@ from typing import TYPE_CHECKING, TypeVar if TYPE_CHECKING: - from typing import Any, AsyncIterator, Callable + from typing import Any, Callable from agents import Agent, Tool, ToolContext @@ -126,7 +125,6 @@ def _create_run_wrapper( ) -> "Callable[..., Any]": """ Wraps the agents.Runner.run methods to - - create and manage a root span for the agent workflow runs. - end the agent invocation span if an `AgentsException` is raised in `run()`. Note agents.Runner.run_sync() is a wrapper around agents.Runner.run(), @@ -150,70 +148,65 @@ async def wrapper(*args: "Any", **kwargs: "Any") -> "Any": else: agent = args[0].clone() - with agent_workflow_span(agent) as workflow_span: - # Set conversation ID on workflow span early so it's captured even on errors - conversation_id = kwargs.get("conversation_id") - if conversation_id: - agent._sentry_conversation_id = conversation_id - - workflow_span.set_attribute( - SPANDATA.GEN_AI_CONVERSATION_ID, conversation_id - ) - - if "starting_agent" in kwargs: - kwargs["starting_agent"] = agent - else: - args = (agent, *args[1:]) - - try: - run_result = await original_func(*args, **kwargs) - except AgentsException as exc: - exc_info = sys.exc_info() - with capture_internal_exceptions(): - _capture_exception(exc) - - context_wrapper = getattr(exc.run_data, "context_wrapper", None) - if context_wrapper is not None: - invoke_agent_span = getattr( - context_wrapper, "_sentry_agent_span", None + # Set conversation ID on workflow span early so it's captured even on errors + conversation_id = kwargs.get("conversation_id") + if conversation_id: + agent._sentry_conversation_id = conversation_id + + if "starting_agent" in kwargs: + kwargs["starting_agent"] = agent + else: + args = (agent, *args[1:]) + + try: + run_result = await original_func(*args, **kwargs) + except AgentsException as exc: + exc_info = sys.exc_info() + with capture_internal_exceptions(): + _capture_exception(exc) + + context_wrapper = getattr(exc.run_data, "context_wrapper", None) + if context_wrapper is not None: + invoke_agent_span = getattr( + context_wrapper, "_sentry_agent_span", None + ) + + if ( + invoke_agent_span is not None + and invoke_agent_span.end_timestamp is None + ): + update_invoke_agent_span( + span=invoke_agent_span, + agent=agent, ) - if ( - invoke_agent_span is not None - and invoke_agent_span.end_timestamp is None - ): - update_invoke_agent_span( - span=invoke_agent_span, - agent=agent, - ) - - invoke_agent_span.__exit__(*exc_info) - delattr(context_wrapper, "_sentry_agent_span") - reraise(*exc_info) - except Exception as exc: - exc_info = sys.exc_info() - with capture_internal_exceptions(): - # Invoke agent span is not finished in this case. - # This is much less likely to occur than other cases because - # AgentRunner.run() is "just" a while loop around _run_single_turn. - _capture_exception(exc) - reraise(*exc_info) - - invoke_agent_span = getattr( - run_result.context_wrapper, "_sentry_agent_span", None - ) - if not invoke_agent_span: - return run_result - - update_invoke_agent_span( - span=invoke_agent_span, - agent=agent, - ) - - invoke_agent_span.__exit__(None, None, None) - delattr(run_result.context_wrapper, "_sentry_agent_span") + invoke_agent_span.__exit__(*exc_info) + delattr(context_wrapper, "_sentry_agent_span") + reraise(*exc_info) + except Exception as exc: + exc_info = sys.exc_info() + with capture_internal_exceptions(): + # Invoke agent span is not finished in this case. + # This is much less likely to occur than other cases because + # AgentRunner.run() is "just" a while loop around _run_single_turn. + _capture_exception(exc) + reraise(*exc_info) + + invoke_agent_span = getattr( + run_result.context_wrapper, "_sentry_agent_span", None + ) + if not invoke_agent_span: return run_result + update_invoke_agent_span( + span=invoke_agent_span, + agent=agent, + ) + + invoke_agent_span.__exit__(None, None, None) + delattr(run_result.context_wrapper, "_sentry_agent_span") + return run_result + return wrapper @@ -221,17 +214,7 @@ def _create_run_streamed_wrapper( original_func: "Callable[..., Any]", ) -> "Callable[..., Any]": """ - Wraps the agents.Runner.run_streamed method to - - create a root span for streaming agent workflow runs. - - end the workflow span if and only if the response stream is consumed or cancelled. - - Unlike run(), run_streamed() returns immediately with a RunResultStreaming object - while execution continues in a background task. The workflow span must stay open - throughout the streaming operation and close when streaming completes or is abandoned. - - Note: We don't use isolation_scope() here because it uses context variables that - cannot span async boundaries (the __enter__ and __exit__ would be called from - different async contexts, causing ValueError). + Wraps the agents.Runner.run_streamed method to inject run hooks. """ @wraps(original_func) @@ -247,19 +230,6 @@ def wrapper(*args: "Any", **kwargs: "Any") -> "Any": if conversation_id: agent._sentry_conversation_id = conversation_id - # Start workflow span immediately (before run_streamed returns) - workflow_span = agent_workflow_span(agent) - workflow_span.__enter__() - - # Set conversation ID on workflow span early so it's captured even on errors - if conversation_id: - workflow_span.set_attribute( - SPANDATA.GEN_AI_CONVERSATION_ID, conversation_id - ) - - # Store span on agent for cleanup - agent._sentry_workflow_span = workflow_span - if "starting_agent" in kwargs: kwargs["starting_agent"] = agent else: @@ -276,45 +246,9 @@ def wrapper(*args: "Any", **kwargs: "Any") -> "Any": # Call original function to get RunResultStreaming run_result = original_func(*args, **kwargs) except Exception as exc: - # If run_streamed itself fails (not the background task), clean up immediately - workflow_span.__exit__(*sys.exc_info()) _capture_exception(exc) raise - def _close_workflow_span() -> None: - if hasattr(agent, "_sentry_workflow_span"): - workflow_span.__exit__(*sys.exc_info()) - delattr(agent, "_sentry_workflow_span") - - if hasattr(run_result, "stream_events"): - original_stream_events = run_result.stream_events - - @wraps(original_stream_events) - async def wrapped_stream_events( - *stream_args: "Any", **stream_kwargs: "Any" - ) -> "AsyncIterator[Any]": - try: - async for event in original_stream_events( - *stream_args, **stream_kwargs - ): - yield event - finally: - _close_workflow_span() - - run_result.stream_events = wrapped_stream_events - - if hasattr(run_result, "cancel"): - original_cancel = run_result.cancel - - @wraps(original_cancel) - def wrapped_cancel(*cancel_args: "Any", **cancel_kwargs: "Any") -> "Any": - try: - return original_cancel(*cancel_args, **cancel_kwargs) - finally: - _close_workflow_span() - - run_result.cancel = wrapped_cancel - return run_result return wrapper diff --git a/sentry_sdk/integrations/openai_agents/spans/__init__.py b/sentry_sdk/integrations/openai_agents/spans/__init__.py index cccc667d31..2675edab34 100644 --- a/sentry_sdk/integrations/openai_agents/spans/__init__.py +++ b/sentry_sdk/integrations/openai_agents/spans/__init__.py @@ -1,4 +1,3 @@ -from .agent_workflow import agent_workflow_span # noqa: F401 from .ai_client import ai_client_context, update_ai_client_span # noqa: F401 from .execute_tool import execute_tool_span, update_execute_tool_span # noqa: F401 from .handoff import handoff_span # noqa: F401 diff --git a/sentry_sdk/integrations/openai_agents/spans/agent_workflow.py b/sentry_sdk/integrations/openai_agents/spans/agent_workflow.py deleted file mode 100644 index 3748f82e2a..0000000000 --- a/sentry_sdk/integrations/openai_agents/spans/agent_workflow.py +++ /dev/null @@ -1,16 +0,0 @@ -from typing import TYPE_CHECKING - -import sentry_sdk - -from ..consts import SPAN_ORIGIN - -if TYPE_CHECKING: - import agents - - -def agent_workflow_span( - agent: "agents.Agent", -) -> "sentry_sdk.traces.Span": - return sentry_sdk.traces.start_span( - name=f"{agent.name} workflow", attributes={"sentry.origin": SPAN_ORIGIN} - ) diff --git a/tests/integrations/openai_agents/test_openai_agents.py b/tests/integrations/openai_agents/test_openai_agents.py index cf67c09147..31a1cb8e48 100644 --- a/tests/integrations/openai_agents/test_openai_agents.py +++ b/tests/integrations/openai_agents/test_openai_agents.py @@ -465,9 +465,6 @@ async def test_agent_invocation_span_no_pii( span for span in spans if span["attributes"]["sentry.op"] == OP.GEN_AI_CHAT ) - assert spans[2]["name"] == "test_agent workflow" - assert spans[2]["attributes"]["sentry.origin"] == "auto.ai.openai_agents" - assert invoke_agent_span["name"] == "invoke_agent test_agent" assert SPANDATA.GEN_AI_SYSTEM_INSTRUCTIONS not in invoke_agent_span["attributes"] @@ -1193,10 +1190,7 @@ async def test_agent_invocation_span( sentry_sdk.flush() spans = [item.payload for item in items] - ai_client_span, invoke_agent_span, workflow_span = spans - - assert workflow_span["name"] == "test_agent workflow" - assert workflow_span["attributes"]["sentry.origin"] == "auto.ai.openai_agents" + ai_client_span, invoke_agent_span = spans assert invoke_agent_span["name"] == "invoke_agent test_agent" @@ -1324,9 +1318,6 @@ def test_agent_invocation_span_sync_no_pii( sentry_sdk.flush() spans = [item.payload for item in items] - assert spans[2]["name"] == "test_agent workflow" - assert spans[2]["attributes"]["sentry.origin"] == "auto.ai.openai_agents" - invoke_agent_span = next( span for span in spans @@ -1594,10 +1585,7 @@ def test_agent_invocation_span_sync( sentry_sdk.flush() spans = [item.payload for item in items] - ai_client_span, invoke_agent_span, workflow_span = spans - - assert workflow_span["name"] == "test_agent workflow" - assert workflow_span["attributes"]["sentry.origin"] == "auto.ai.openai_agents" + ai_client_span, invoke_agent_span = spans assert invoke_agent_span["name"] == "invoke_agent test_agent" assert invoke_agent_span["attributes"]["gen_ai.operation.name"] == "invoke_agent" @@ -1985,9 +1973,6 @@ async def test_tool_execution_span( sentry_sdk.flush() spans = [item.payload for item in items] - assert spans[4]["name"] == "test_agent workflow" - assert spans[4]["attributes"]["sentry.origin"] == "auto.ai.openai_agents" - agent_span = next( span for span in spans @@ -2858,10 +2843,7 @@ async def test_model_behavior_error( ( ai_client_span1, agent_span, - workflow_span, ) = spans - assert workflow_span["name"] == "test_agent workflow" - assert workflow_span["attributes"]["sentry.origin"] == "auto.ai.openai_agents" assert agent_span["name"] == "invoke_agent test_agent" assert agent_span["attributes"]["sentry.origin"] == "auto.ai.openai_agents" @@ -2908,10 +2890,10 @@ async def test_run_error_handling( sentry_sdk.flush() spans = [item.payload for item in items if item.type == "span"] - (ai_client_span, invoke_agent_span, workflow_span) = spans - - assert workflow_span["name"] == "test_agent workflow" - assert workflow_span["attributes"]["sentry.origin"] == "auto.ai.openai_agents" + ( + ai_client_span, + invoke_agent_span, + ) = spans assert invoke_agent_span["name"] == "invoke_agent test_agent" assert invoke_agent_span["attributes"]["sentry.origin"] == "auto.ai.openai_agents" @@ -2956,10 +2938,10 @@ async def test_run_streamed_error_handling( sentry_sdk.flush() spans = [item.payload for item in items if item.type == "span"] - (ai_client_span, invoke_agent_span, workflow_span) = spans - - assert workflow_span["name"] == "test_agent workflow" - assert workflow_span["attributes"]["sentry.origin"] == "auto.ai.openai_agents" + ( + ai_client_span, + invoke_agent_span, + ) = spans assert invoke_agent_span["name"] == "invoke_agent test_agent" assert invoke_agent_span["attributes"]["sentry.origin"] == "auto.ai.openai_agents" @@ -3069,58 +3051,6 @@ async def test_span_status_error( spans = [item.payload for item in items if item.type == "span"] assert spans[0]["status"] == "error" - assert spans[2]["is_segment"] is True - assert spans[2]["status"] == "error" - - -@pytest.mark.asyncio -async def test_multiple_agents_asyncio( - sentry_init, - capture_items, - test_agent, - nonstreaming_responses_model_response, - get_model_response, -): - """ - Test that multiple agents can be run at the same time in asyncio tasks - without interfering with each other. - """ - client = AsyncOpenAI(api_key="test-key") - model = OpenAIResponsesModel(model="gpt-4", openai_client=client) - agent = test_agent.clone(model=model) - - response = get_model_response( - nonstreaming_responses_model_response, serialize_pydantic=True - ) - with patch.object( - agent.model._client._client, - "send", - return_value=response, - ) as _: - sentry_init( - integrations=[OpenAIAgentsIntegration()], - disabled_integrations=[StdlibIntegration], - traces_sample_rate=1.0, - ) - - items = capture_items("span") - - async def run(): - await agents.Runner.run( - starting_agent=agent, - input="Test input", - run_config=test_run_config, - ) - - await asyncio.gather(*[run() for _ in range(3)]) - - sentry_sdk.flush() - spans = [item.payload for item in items] - - assert spans[2]["name"] == "test_agent workflow" - assert spans[5]["name"] == "test_agent workflow" - assert spans[8]["name"] == "test_agent workflow" - # Test input messages with mixed roles including "ai" @pytest.mark.parametrize( @@ -3916,8 +3846,6 @@ async def test_conversation_id_on_all_spans( span for span in spans if span["attributes"]["sentry.op"] == OP.GEN_AI_CHAT ) - assert spans[2]["attributes"]["gen_ai.conversation.id"] == "conv_test_123" - # Verify invoke_agent span has conversation_id assert invoke_agent_span["attributes"]["gen_ai.conversation.id"] == "conv_test_123" @@ -4051,224 +3979,3 @@ def simple_tool(message: str) -> str: assert tool_span is not None # Tool span should have the conversation_id passed to Runner.run() assert tool_span["attributes"]["gen_ai.conversation.id"] == "conv_tool_test_456" - - # Workflow span should have the same conversation_id - workflow_span = spans[4] - assert workflow_span["is_segment"] is True - - assert workflow_span["attributes"]["gen_ai.conversation.id"] == "conv_tool_test_456" - - -@pytest.mark.asyncio -async def test_no_conversation_id_when_not_provided( - sentry_init, - capture_items, - test_agent, - nonstreaming_responses_model_response, - get_model_response, -): - """ - Test that gen_ai.conversation.id is not set when not passed to Runner.run(). - """ - - client = AsyncOpenAI(api_key="test-key") - model = OpenAIResponsesModel(model="gpt-4", openai_client=client) - agent = test_agent.clone(model=model) - - response = get_model_response( - nonstreaming_responses_model_response, serialize_pydantic=True - ) - with patch.object( - agent.model._client._client, - "send", - return_value=response, - ) as _: - sentry_init( - integrations=[OpenAIAgentsIntegration()], - disabled_integrations=[StdlibIntegration], - traces_sample_rate=1.0, - ) - - items = capture_items("span") - - # Don't pass conversation_id - result = await agents.Runner.run( - agent, "Test input", run_config=test_run_config - ) - - assert result is not None - - sentry_sdk.flush() - spans = [item.payload for item in items] - - workflow_span = spans[2] - assert workflow_span["is_segment"] is True - - invoke_agent_span = next( - span - for span in spans - if span["attributes"]["sentry.op"] == OP.GEN_AI_INVOKE_AGENT - ) - ai_client_span = next( - span for span in spans if span["attributes"]["sentry.op"] == OP.GEN_AI_CHAT - ) - - # Verify conversation_id is NOT set on any spans - assert "gen_ai.conversation.id" not in workflow_span.get("attributes", {}) - assert "gen_ai.conversation.id" not in invoke_agent_span.get("attributes", {}) - assert "gen_ai.conversation.id" not in ai_client_span.get("attributes", {}) - - -@pytest.mark.asyncio -async def test_runner_run_with_starting_agent_kwarg( - sentry_init, - capture_items, - test_agent, - nonstreaming_responses_model_response, - get_model_response, -): - """Runner.run(starting_agent=agent, input=...) must not crash. - - Regression test for https://github.com/getsentry/sentry-python/issues/6418 - """ - client = AsyncOpenAI(api_key="test-key") - model = OpenAIResponsesModel(model="gpt-4", openai_client=client) - agent = test_agent.clone(model=model) - - response = get_model_response( - nonstreaming_responses_model_response, serialize_pydantic=True - ) - - with patch.object( - agent.model._client._client, - "send", - return_value=response, - ): - sentry_init( - integrations=[OpenAIAgentsIntegration()], - traces_sample_rate=1.0, - ) - - items = capture_items("span") - - result = await agents.run.DEFAULT_AGENT_RUNNER.run( - starting_agent=agent, - input="Test input", - run_config=test_run_config, - ) - - assert result is not None - assert result.final_output == "Hello, how can I help you?" - - sentry_sdk.flush() - spans = [item.payload for item in items] - assert any(span["name"] == "test_agent workflow" for span in spans) - - -@pytest.mark.asyncio -async def test_runner_run_streamed_with_starting_agent_kwarg( - sentry_init, - capture_items, - test_agent, - async_iterator, - server_side_event_chunks, - get_model_response, -): - """Runner.run_streamed(starting_agent=agent, input=...) must not crash. - - Regression test for https://github.com/getsentry/sentry-python/issues/6418 - """ - client = AsyncOpenAI(api_key="test-key") - model = OpenAIResponsesModel(model="gpt-4", openai_client=client) - agent = test_agent.clone(model=model) - - request_headers = {"X-Stainless-Raw-Response": "stream"} - - response = get_model_response( - async_iterator( - server_side_event_chunks( - [ - ResponseCreatedEvent( - response=Response( - id="chat-id", - output=[], - parallel_tool_calls=False, - tool_choice="none", - tools=[], - created_at=10000000, - model="gpt-4", - object="response", - ), - type="response.created", - sequence_number=0, - ), - ResponseCompletedEvent( - response=Response( - id="chat-id", - output=[ - ResponseOutputMessage( - id="message-id", - content=[ - ResponseOutputText( - annotations=[], - text="Hello, how can I help you?", - type="output_text", - ), - ], - role="assistant", - status="completed", - type="message", - ), - ], - parallel_tool_calls=False, - tool_choice="none", - tools=[], - created_at=10000000, - model="gpt-4", - object="response", - usage=ResponseUsage( - input_tokens=10, - input_tokens_details=InputTokensDetails( - cached_tokens=0, - cache_write_tokens=0, - ), - output_tokens=20, - output_tokens_details=OutputTokensDetails( - reasoning_tokens=5, - ), - total_tokens=30, - ), - ), - type="response.completed", - sequence_number=1, - ), - ] - ) - ), - request_headers=request_headers, - ) - - with patch.object( - agent.model._client._client, - "send", - return_value=response, - ): - sentry_init( - integrations=[OpenAIAgentsIntegration()], - traces_sample_rate=1.0, - ) - - items = capture_items("span") - - result = agents.run.DEFAULT_AGENT_RUNNER.run_streamed( - starting_agent=agent, - input="Test input", - run_config=test_run_config, - ) - - async for _event in result.stream_events(): - pass - - sentry_sdk.flush() - spans = [item.payload for item in items] - assert any(span["name"] == "test_agent workflow" for span in spans)