diff --git a/src/agents/tracing/processors.py b/src/agents/tracing/processors.py index b61f3e7976..b6f28c1aac 100644 --- a/src/agents/tracing/processors.py +++ b/src/agents/tracing/processors.py @@ -534,6 +534,16 @@ def close(self): """Close the underlying HTTP client.""" self._client.close() + @property + def is_shut_down(self) -> bool: + """Whether this exporter has been shut down. + + Shutdown is terminal. The signal that abandons retry backoff is never cleared, so a + shut-down exporter gives up on the first transient failure instead of retrying. Reuse + one and every batch that hits a 5xx is dropped; build a fresh exporter instead. + """ + return self._shutdown_event.is_set() + def _request_shutdown(self) -> None: self._shutdown_event.set() @@ -620,6 +630,11 @@ def on_span_end(self, span: Span[Any]) -> None: except queue.Full: logger.warning("Queue is full, dropping span.") + @property + def is_shut_down(self) -> bool: + """Whether ``shutdown`` has been called. Terminal: a shut-down processor stays down.""" + return self._shutdown_event.is_set() + def shutdown(self, timeout: float | None = None): """ Called when the application stops. We signal our thread to stop, then join it. @@ -724,21 +739,57 @@ def _export_batches(self, deadline: float | None = None): _global_lock = threading.Lock() +def _replacement_exporter(exporter: BackendSpanExporter) -> BackendSpanExporter: + """A live exporter configured exactly like the shut-down one it replaces. + + Everything a caller can configure -- the API key set through + `set_tracing_export_api_key`, the organization and project, the endpoint, and the + retry schedule -- is carried forward, so recovering from a shutdown restores the + configured exporter rather than one derived from the environment alone. + """ + return BackendSpanExporter( + api_key=exporter._api_key, + organization=exporter._organization, + project=exporter._project, + endpoint=exporter.endpoint, + max_retries=exporter.max_retries, + base_delay=exporter.base_delay, + max_delay=exporter.max_delay, + ) + + +def _replace_shut_down_defaults() -> None: + """Swap out whichever cached default shutdown has made terminal. + + An exporter's shutdown signal -- the one that abandons retry backoff -- is never + cleared, so exporting through it again looks like it worked while dropping every + batch that hits a transient failure. Replace it, carrying its configuration over. + The processor it belonged to goes with it, since it exports through the dead one. + + A shutdown with no timeout never signals the exporter, so only the processor is + terminal there and the live exporter is kept and reused. + + Callers must hold `_global_lock`. + """ + global _global_exporter + global _global_processor + + if _global_exporter is not None and _global_exporter.is_shut_down: + _global_exporter = _replacement_exporter(_global_exporter) + _global_processor = None + elif _global_processor is not None and _global_processor.is_shut_down: + _global_processor = None + + def default_exporter() -> BackendSpanExporter: """The default exporter, which exports traces and spans to the backend in batches.""" global _global_exporter - exporter = _global_exporter - if exporter is not None: - return exporter - with _global_lock: - exporter = _global_exporter - if exporter is None: - exporter = BackendSpanExporter() - _global_exporter = exporter - - return exporter + _replace_shut_down_defaults() + if _global_exporter is None: + _global_exporter = BackendSpanExporter() + return _global_exporter def default_processor() -> BatchTraceProcessor: @@ -746,18 +797,10 @@ def default_processor() -> BatchTraceProcessor: global _global_exporter global _global_processor - processor = _global_processor - if processor is not None: - return processor - with _global_lock: - processor = _global_processor - if processor is None: - exporter = _global_exporter - if exporter is None: - exporter = BackendSpanExporter() - _global_exporter = exporter - processor = BatchTraceProcessor(exporter) - _global_processor = processor - - return processor + _replace_shut_down_defaults() + if _global_processor is None: + if _global_exporter is None: + _global_exporter = BackendSpanExporter() + _global_processor = BatchTraceProcessor(_global_exporter) + return _global_processor diff --git a/src/agents/tracing/provider.py b/src/agents/tracing/provider.py index b0e10b0bd2..850de2ebdb 100644 --- a/src/agents/tracing/provider.py +++ b/src/agents/tracing/provider.py @@ -114,6 +114,16 @@ def set_processors(self, processors: list[TracingProcessor]): with self._lock: self._processors = tuple(processors) + def _replace_processor(self, old: TracingProcessor, new: TracingProcessor) -> None: + """Swap one registered processor for another, keeping registration order. + + Lets a shut-down default processor be recovered without rebuilding the provider, + which would discard every other processor registered on it. A no-op when `old` is + no longer registered, which is what `set_trace_processors` leaves behind. + """ + with self._lock: + self._processors = tuple(new if p is old else p for p in self._processors) + def on_trace_start(self, trace: Trace) -> None: """ Called when a trace is started. @@ -317,6 +327,10 @@ def set_processors(self, processors: list[TracingProcessor]): """ self._multi_processor.set_processors(processors) + def _replace_processor(self, old: TracingProcessor, new: TracingProcessor) -> None: + """Swap one registered processor for another, keeping registration order.""" + self._multi_processor._replace_processor(old, new) + def get_current_trace(self) -> Trace | None: """ Returns the currently active trace, if any. diff --git a/src/agents/tracing/setup.py b/src/agents/tracing/setup.py index 0ec72de239..c001d7719f 100644 --- a/src/agents/tracing/setup.py +++ b/src/agents/tracing/setup.py @@ -5,15 +5,61 @@ from typing import TYPE_CHECKING if TYPE_CHECKING: + from .processor_interface import TracingProcessor from .provider import TraceProvider _DEFAULT_SHUTDOWN_TIMEOUT = 5.0 GLOBAL_TRACE_PROVIDER: TraceProvider | None = None _GLOBAL_TRACE_PROVIDER_LOCK = threading.Lock() _SHUTDOWN_HANDLER_REGISTERED = False +# The default processor wired into a provider we bootstrapped ourselves, kept so a +# terminal shutdown can be spotted and re-initialized. None whenever the provider came +# from `set_trace_provider`: that lifecycle belongs to whoever set it. +_DEFAULT_PROCESSOR: TracingProcessor | None = None +# Set before the atexit shutdown runs, so a span closed during interpreter teardown +# cannot resurrect the stack and build a fresh HTTP client on the way out. +_ATEXIT_SHUTDOWN_STARTED = False + + +def _default_processor_is_terminal() -> bool: + """Whether the default processor we bootstrapped has been shut down. + + Shutdown is terminal for it, so the next use recovers instead of feeding spans to a + processor that will never drain them again, or applying a freshly configured API key + to an exporter nothing exports through. + """ + if _ATEXIT_SHUTDOWN_STARTED: + return False + return getattr(_DEFAULT_PROCESSOR, "is_shut_down", False) is True + + +def _recover_default_processor(provider: TraceProvider) -> None: + """Swap the shut-down default processor for a live one, in place. + + Only that one processor is replaced. Rebuilding the provider instead would silently + discard everything the caller configured on it -- `set_tracing_disabled`, and every + processor added through `add_trace_processor` -- none of which shutdown invalidated. + + Callers must hold `_GLOBAL_TRACE_PROVIDER_LOCK`. + """ + global _DEFAULT_PROCESSOR + + from .processors import default_processor + from .provider import DefaultTraceProvider + + stale = _DEFAULT_PROCESSOR + fresh = default_processor() + if fresh is stale: + return + if stale is not None and isinstance(provider, DefaultTraceProvider): + provider._replace_processor(stale, fresh) + _DEFAULT_PROCESSOR = fresh def _shutdown_global_trace_provider() -> None: + global _ATEXIT_SHUTDOWN_STARTED + + _ATEXIT_SHUTDOWN_STARTED = True provider = GLOBAL_TRACE_PROVIDER if provider is not None: from .provider import DefaultTraceProvider @@ -28,9 +74,11 @@ def set_trace_provider(provider: TraceProvider) -> None: """Set the global trace provider used by tracing utilities.""" global GLOBAL_TRACE_PROVIDER global _SHUTDOWN_HANDLER_REGISTERED + global _DEFAULT_PROCESSOR with _GLOBAL_TRACE_PROVIDER_LOCK: GLOBAL_TRACE_PROVIDER = provider + _DEFAULT_PROCESSOR = None if not _SHUTDOWN_HANDLER_REGISTERED: atexit.register(_shutdown_global_trace_provider) _SHUTDOWN_HANDLER_REGISTERED = True @@ -44,9 +92,10 @@ def get_trace_provider() -> TraceProvider: """ global GLOBAL_TRACE_PROVIDER global _SHUTDOWN_HANDLER_REGISTERED + global _DEFAULT_PROCESSOR provider = GLOBAL_TRACE_PROVIDER - if provider is not None: + if provider is not None and not _default_processor_is_terminal(): return provider with _GLOBAL_TRACE_PROVIDER_LOCK: @@ -55,9 +104,13 @@ def get_trace_provider() -> TraceProvider: from .processors import default_processor from .provider import DefaultTraceProvider + processor = default_processor() provider = DefaultTraceProvider() - provider.register_processor(default_processor()) + provider.register_processor(processor) GLOBAL_TRACE_PROVIDER = provider + _DEFAULT_PROCESSOR = processor + elif _default_processor_is_terminal(): + _recover_default_processor(provider) if not _SHUTDOWN_HANDLER_REGISTERED: atexit.register(_shutdown_global_trace_provider) diff --git a/tests/test_trace_processor.py b/tests/test_trace_processor.py index 07e975ccb9..ca506d8380 100644 --- a/tests/test_trace_processor.py +++ b/tests/test_trace_processor.py @@ -12,7 +12,12 @@ import pytest import agents._debug as _debug -from agents.tracing import flush_traces, get_trace_provider +from agents.tracing import ( + flush_traces, + get_trace_provider, + set_tracing_export_api_key, + setup as tracing_setup, +) from agents.tracing.processor_interface import TracingExporter, TracingProcessor from agents.tracing.processors import BackendSpanExporter, BatchTraceProcessor, ConsoleSpanExporter from agents.tracing.provider import DefaultTraceProvider, TraceProvider @@ -20,6 +25,8 @@ from agents.tracing.spans import Span, SpanImpl from agents.tracing.traces import Trace, TraceImpl +from .testing_processor import SpanProcessorForTests + def get_span(processor: TracingProcessor) -> SpanImpl[AgentSpanData]: """Create a minimal agent span for testing processors.""" @@ -1354,3 +1361,279 @@ def test_truncate_string_for_json_limit_handles_escape_heavy_input(): assert truncated.endswith(exporter._OPENAI_TRACING_STRING_TRUNCATION_SUFFIX) assert exporter._value_json_size_bytes(truncated) <= max_bytes exporter.close() + + +@pytest.fixture +def restore_tracing_defaults(): + """Isolate the module-level default exporter/processor/provider around a test.""" + from agents.tracing import processors as tracing_processors + + saved = ( + tracing_processors._global_exporter, + tracing_processors._global_processor, + tracing_setup.GLOBAL_TRACE_PROVIDER, + tracing_setup._DEFAULT_PROCESSOR, + tracing_setup._ATEXIT_SHUTDOWN_STARTED, + ) + tracing_processors._global_exporter = None + tracing_processors._global_processor = None + tracing_setup.GLOBAL_TRACE_PROVIDER = None + tracing_setup._DEFAULT_PROCESSOR = None + tracing_setup._ATEXIT_SHUTDOWN_STARTED = False + try: + yield tracing_processors + finally: + exporter = tracing_processors._global_exporter + if exporter is not None: + exporter.close() + ( + tracing_processors._global_exporter, + tracing_processors._global_processor, + tracing_setup.GLOBAL_TRACE_PROVIDER, + tracing_setup._DEFAULT_PROCESSOR, + tracing_setup._ATEXIT_SHUTDOWN_STARTED, + ) = saved + + +def test_shut_down_flags_are_terminal(): + exporter = BackendSpanExporter(api_key="test_key") + processor = BatchTraceProcessor(exporter=exporter) + + assert not exporter.is_shut_down + assert not processor.is_shut_down + + processor.shutdown(timeout=1.0) + + assert processor.is_shut_down + assert exporter.is_shut_down + exporter.close() + + +def test_shutdown_without_timeout_leaves_the_exporter_usable(): + """No timeout means no retry-abandoning signal, so only the processor goes down.""" + exporter = BackendSpanExporter(api_key="test_key") + processor = BatchTraceProcessor(exporter=exporter) + + processor.shutdown() + + assert processor.is_shut_down + assert not exporter.is_shut_down + exporter.close() + + +def test_default_pair_is_replaced_after_shutdown(restore_tracing_defaults): + tracing_processors = restore_tracing_defaults + + first_processor = tracing_processors.default_processor() + first_exporter = tracing_processors.default_exporter() + assert first_processor._exporter is first_exporter + + first_processor.shutdown(timeout=1.0) + first_exporter.close() + + second_exporter = tracing_processors.default_exporter() + second_processor = tracing_processors.default_processor() + + assert second_exporter is not first_exporter + assert second_processor is not first_processor + assert second_processor._exporter is second_exporter + assert not second_exporter.is_shut_down + assert not second_processor.is_shut_down + + +def test_only_the_processor_is_replaced_when_the_exporter_is_still_live( + restore_tracing_defaults, +): + """A no-timeout shutdown never signals the exporter, so it is kept rather than leaked.""" + tracing_processors = restore_tracing_defaults + + first_processor = tracing_processors.default_processor() + first_exporter = tracing_processors.default_exporter() + + first_processor.shutdown() + assert not first_exporter.is_shut_down + + assert tracing_processors.default_exporter() is first_exporter + assert tracing_processors.default_processor() is not first_processor + assert tracing_processors.default_processor()._exporter is first_exporter + + +def test_default_pair_is_stable_while_live(restore_tracing_defaults): + tracing_processors = restore_tracing_defaults + + assert tracing_processors.default_exporter() is tracing_processors.default_exporter() + assert tracing_processors.default_processor() is tracing_processors.default_processor() + + +def test_default_exporter_does_not_build_a_processor(restore_tracing_defaults): + """Keep the processor lazy: asking for the exporter must not create threading primitives.""" + tracing_processors = restore_tracing_defaults + + tracing_processors.default_exporter() + + assert tracing_processors._global_processor is None + + +def test_fresh_default_exporter_still_retries_transient_failures(restore_tracing_defaults): + """Regression for #4683: a reused exporter used to give up on the first 5xx forever.""" + tracing_processors = restore_tracing_defaults + + with patch("httpx2.Client") as mock_client: + response = MagicMock() + response.status_code = 504 + mock_client.return_value.post.return_value = response + + exporter = tracing_processors.default_exporter() + exporter.set_api_key("test_key") + exporter.max_retries = 3 + exporter.base_delay = 0.0001 + exporter.max_delay = 0.0002 + + first = tracing_processors.default_processor() + first.shutdown(timeout=1.0) + + # A later tracing initialization gets a fresh pair, not the shut-down one. + second_exporter = tracing_processors.default_exporter() + second_exporter.set_api_key("test_key") + second_exporter.max_retries = 3 + second_exporter.base_delay = 0.0001 + second_exporter.max_delay = 0.0002 + second = tracing_processors.default_processor() + + assert second_exporter is not exporter + assert second is not first + + # The shut-down exporter is the broken one: it abandons backoff on the first 5xx. + mock_client.return_value.post.reset_mock() + stale = BatchTraceProcessor(exporter=exporter) + stale._queue.put_nowait(get_span(stale)) + stale.force_flush() + assert mock_client.return_value.post.call_count == 1 + + # The fresh one still retries. + mock_client.return_value.post.reset_mock() + second._queue.put_nowait(get_span(second)) + second.force_flush() + assert mock_client.return_value.post.call_count == 3 # max_retries counts attempts + + +def test_recovery_preserves_everything_configured_on_the_stack(restore_tracing_defaults): + """Recovering from a shutdown must restore the configured stack, not a default one.""" + tracing_processors = restore_tracing_defaults + + provider = tracing_setup.get_trace_provider() + first_exporter = tracing_processors.default_exporter() + first_processor = tracing_processors.default_processor() + + set_tracing_export_api_key("trace_only_key") + first_exporter._organization = "org_x" + first_exporter._project = "proj_x" + first_exporter.endpoint = "https://example.test/ingest" + first_exporter.max_retries = 7 + first_exporter.base_delay = 2.5 + first_exporter.max_delay = 40.0 + cast(Any, provider).set_disabled(True) + mine = SpanProcessorForTests() + provider.register_processor(mine) + + cast(Any, provider).shutdown(timeout=1.0) + first_exporter.close() + + recovered = tracing_setup.get_trace_provider() + + # The provider itself is kept, so nothing configured on it is lost. + assert recovered is provider + assert cast(Any, recovered)._manual_disabled is True + registered = cast(Any, recovered)._multi_processor._processors + assert mine in registered + + # Only the dead default processor is swapped, in place, keeping registration order. + fresh = tracing_processors.default_processor() + assert registered == (fresh, mine) + assert fresh is not first_processor + assert not fresh.is_shut_down + + # The replacement exporter carries every configured value forward. + exporter = fresh._exporter + assert exporter is tracing_processors.default_exporter() + assert exporter is not first_exporter + assert not exporter.is_shut_down + assert exporter.api_key == "trace_only_key" + assert exporter.organization == "org_x" + assert exporter.project == "proj_x" + assert exporter.endpoint == "https://example.test/ingest" + assert (exporter.max_retries, exporter.base_delay, exporter.max_delay) == (7, 2.5, 40.0) + + +def test_recovery_reuses_the_exporter_a_no_timeout_shutdown_left_alive( + restore_tracing_defaults, +): + """No timeout means no shutdown signal on the exporter, so it is kept, not replaced.""" + tracing_processors = restore_tracing_defaults + + provider = tracing_setup.get_trace_provider() + exporter = tracing_processors.default_exporter() + first_processor = tracing_processors.default_processor() + + provider.shutdown() + assert not exporter.is_shut_down + + tracing_setup.get_trace_provider() + fresh = tracing_processors.default_processor() + + assert fresh is not first_processor + assert tracing_processors.default_exporter() is exporter + assert fresh._exporter is exporter + + +def test_live_provider_is_not_replaced(restore_tracing_defaults): + provider = tracing_setup.get_trace_provider() + + assert tracing_setup.get_trace_provider() is provider + + +def test_caller_supplied_provider_is_never_recovered(restore_tracing_defaults): + """`set_trace_provider` hands the lifecycle to the caller, shut down or not.""" + provider = DefaultTraceProvider() + processor = BatchTraceProcessor(exporter=BackendSpanExporter(api_key="test_key")) + provider.set_processors([processor]) + tracing_setup.set_trace_provider(provider) + + provider.shutdown(timeout=1.0) + processor._exporter.close() + + assert tracing_setup.get_trace_provider() is provider + assert cast(Any, provider)._multi_processor._processors == (processor,) + + +def test_set_trace_processors_opt_out_survives_recovery(restore_tracing_defaults): + """Dropping the default processor is the caller's choice; recovery must not undo it.""" + tracing_processors = restore_tracing_defaults + + provider = tracing_setup.get_trace_provider() + stale = tracing_processors.default_processor() + mine = SpanProcessorForTests() + provider.set_processors([mine]) + + cast(Any, provider).shutdown(timeout=1.0) + tracing_processors.default_exporter().close() + + recovered = tracing_setup.get_trace_provider() + + assert recovered is provider + assert cast(Any, recovered)._multi_processor._processors == (mine,) + assert stale not in cast(Any, recovered)._multi_processor._processors + + +def test_atexit_shutdown_does_not_resurrect_the_stack(restore_tracing_defaults): + """Teardown must not build a fresh exporter and HTTP client on the way out.""" + tracing_processors = restore_tracing_defaults + + provider = tracing_setup.get_trace_provider() + exporter = tracing_processors.default_exporter() + + tracing_setup._shutdown_global_trace_provider() + + assert tracing_setup._ATEXIT_SHUTDOWN_STARTED + assert tracing_setup.get_trace_provider() is provider + assert tracing_processors._global_exporter is exporter