From 314e8aec59e74254183c0cdb040836403b63a3b5 Mon Sep 17 00:00:00 2001 From: rajarshidattapy Date: Fri, 28 Aug 2026 10:47:49 +0530 Subject: [PATCH 1/4] fix(tracing): rebuild the default exporter and processor after shutdown `BackendSpanExporter._shutdown_event` is the signal that abandons retry backoff, and it is never cleared. Because `default_exporter()` caches a module-level singleton, a processor shutdown left that signal set on an exporter the SDK would hand straight back out. Any batch exported through it afterwards gave up on the first 5xx instead of backing off, and was dropped with a warning blaming a shutdown that had long since finished. Make shutdown terminal instead of clearing the signal: `default_exporter()` and `default_processor()` discard the cached pair once either half has been shut down, so a later tracing initialization starts over with a fresh exporter and processor. The processor owns the exporter it was handed, so the two go down together -- no ref counting and no cross-processor cancellation protocol. Both classes expose `is_shut_down` so the terminal state is explicit rather than a private-attribute peek. Closes #4683 --- src/agents/tracing/processors.py | 69 +++++++++------ tests/test_trace_processor.py | 140 +++++++++++++++++++++++++++++++ 2 files changed, 185 insertions(+), 24 deletions(-) diff --git a/src/agents/tracing/processors.py b/src/agents/tracing/processors.py index b61f3e7976..fa2cc21771 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,35 @@ def _export_batches(self, deadline: float | None = None): _global_lock = threading.Lock() +def _discard_shut_down_defaults() -> None: + """Drop the cached defaults once shutdown has made them terminal. + + Shutdown is terminal, and the processor owns the exporter it was handed, so the pair + goes down together. Handing the shut-down exporter to a new processor would look like + it worked while silently abandoning every retry backoff -- the first 5xx drops the + batch instead of retrying -- so a later tracing initialization gets a fresh pair. + + Callers must hold ``_global_lock``. + """ + global _global_exporter + global _global_processor + + if (_global_exporter is not None and _global_exporter.is_shut_down) or ( + _global_processor is not None and _global_processor.is_shut_down + ): + _global_exporter = None + _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 + _discard_shut_down_defaults() + if _global_exporter is None: + _global_exporter = BackendSpanExporter() + return _global_exporter def default_processor() -> BatchTraceProcessor: @@ -746,18 +775,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 + _discard_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/tests/test_trace_processor.py b/tests/test_trace_processor.py index 07e975ccb9..e15fc77d93 100644 --- a/tests/test_trace_processor.py +++ b/tests/test_trace_processor.py @@ -1354,3 +1354,143 @@ 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 around a test.""" + from agents.tracing import processors as tracing_processors + + saved = (tracing_processors._global_exporter, tracing_processors._global_processor) + tracing_processors._global_exporter = None + tracing_processors._global_processor = None + 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, + ) = 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_default_pair_is_replaced_when_only_the_processor_shut_down(restore_tracing_defaults): + """A no-timeout shutdown never signals the exporter, but the pair is still terminal.""" + 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 + first_exporter.close() + + assert tracing_processors.default_exporter() is not first_exporter + assert tracing_processors.default_processor() is not first_processor + + +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 From bc86b776255deeb49410fa3a86634aedd0866c03 Mon Sep 17 00:00:00 2001 From: rajarshidattapy Date: Fri, 28 Aug 2026 11:14:08 +0530 Subject: [PATCH 2/4] fix(tracing): re-initialize the global provider with the fresh pair Discarding the cached exporter/processor was only half the story. When tracing had been bootstrapped through `get_trace_provider()`, a timed `provider.shutdown()` left that provider registered with the original processor. The next `default_exporter()` or `set_tracing_export_api_key()` call built a fresh pair, but `get_trace_provider()` still handed back the old provider -- so default traces kept flowing through the shut-down exporter, and a newly configured API key landed on an exporter nothing exported through. Track the default processor wired into a provider we bootstrapped ourselves, and re-initialize the provider alongside the cache when that processor is shut down. Shutdown stays terminal for the whole default stack. A provider supplied through `set_trace_provider` is left alone: its lifecycle belongs to whoever set it. Guard the atexit path so a span closed during interpreter teardown cannot resurrect the stack and build a fresh HTTP client on the way out. --- src/agents/tracing/setup.py | 35 +++++++++++++-- tests/test_trace_processor.py | 83 +++++++++++++++++++++++++++++++++-- 2 files changed, 112 insertions(+), 6 deletions(-) diff --git a/src/agents/tracing/setup.py b/src/agents/tracing/setup.py index 0ec72de239..f71c9e5a71 100644 --- a/src/agents/tracing/setup.py +++ b/src/agents/tracing/setup.py @@ -5,15 +5,39 @@ 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_stack_is_terminal() -> bool: + """Whether the bootstrapped provider's default processor has been shut down. + + Shutdown is terminal for the whole default stack -- provider, processor and exporter + alike -- so the next use re-initializes 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 _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 +52,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,20 +70,23 @@ 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_stack_is_terminal(): return provider with _GLOBAL_TRACE_PROVIDER_LOCK: provider = GLOBAL_TRACE_PROVIDER - if provider is None: + if provider is None or _default_stack_is_terminal(): 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 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 e15fc77d93..a5c8b2ce55 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 @@ -1358,12 +1363,21 @@ def test_truncate_string_for_json_limit_handles_escape_heavy_input(): @pytest.fixture def restore_tracing_defaults(): - """Isolate the module-level default exporter/processor around a test.""" + """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) + 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: @@ -1373,6 +1387,9 @@ def restore_tracing_defaults(): ( tracing_processors._global_exporter, tracing_processors._global_processor, + tracing_setup.GLOBAL_TRACE_PROVIDER, + tracing_setup._DEFAULT_PROCESSOR, + tracing_setup._ATEXIT_SHUTDOWN_STARTED, ) = saved @@ -1494,3 +1511,63 @@ def test_fresh_default_exporter_still_retries_transient_failures(restore_tracing 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_shut_down_provider_reinitializes_with_the_fresh_pair(restore_tracing_defaults): + """A stale provider must not keep exporting through the shut-down pair.""" + tracing_processors = restore_tracing_defaults + + first_provider = tracing_setup.get_trace_provider() + first_processor = tracing_processors.default_processor() + first_exporter = tracing_processors.default_exporter() + assert tracing_setup._DEFAULT_PROCESSOR is first_processor + + first_provider.shutdown(timeout=1.0) + first_exporter.close() + + # Reconfiguring must land on the exporter the live provider actually exports through. + set_tracing_export_api_key("fresh_key") + + second_provider = tracing_setup.get_trace_provider() + assert second_provider is not first_provider + + registered = cast(Any, second_provider)._multi_processor._processors + assert len(registered) == 1 + assert registered[0] is tracing_processors.default_processor() + assert registered[0] is not first_processor + assert registered[0]._exporter is tracing_processors.default_exporter() + assert registered[0]._exporter is not first_exporter + assert registered[0]._exporter.api_key == "fresh_key" + + +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_replaced(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 + + +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 From 069ed650779545cbff41938de2ba4b5ff0bcad80 Mon Sep 17 00:00:00 2001 From: rajarshidattapy Date: Fri, 28 Aug 2026 11:36:51 +0530 Subject: [PATCH 3/4] fix(tracing): recover in place, preserving everything the caller configured Rebuilding objects to recover from a shutdown meant every piece of caller-set configuration became something the recovery path had to remember to copy, and it was not copying any of it. Two structural fixes remove the whole class: Keep the provider. Only the one dead default processor is swapped, in place, via `SynchronousMultiTracingProcessor._replace_processor`. `set_tracing_disabled`, the cached env flag, and every processor added through `add_trace_processor` survive untouched, and registration order is preserved. A caller who dropped the default processor with `set_trace_processors` does not get it back. Carry the exporter's configuration. `_replacement_exporter` copies the API key set through `set_tracing_export_api_key` along with the organization, project, endpoint and retry schedule, so recovery restores the configured exporter rather than one derived from the environment. An application relying on a trace-only key no longer silently stops exporting after a shutdown. The exporter is only replaced when it is itself shut down. A shutdown with no timeout never signals it, so it is now reused rather than dropped, which also stops the recovery path leaking a live HTTP client. --- src/agents/tracing/processors.py | 48 ++++++++++---- src/agents/tracing/provider.py | 14 ++++ src/agents/tracing/setup.py | 40 ++++++++--- tests/test_trace_processor.py | 110 ++++++++++++++++++++++++------- 4 files changed, 169 insertions(+), 43 deletions(-) diff --git a/src/agents/tracing/processors.py b/src/agents/tracing/processors.py index fa2cc21771..b6f28c1aac 100644 --- a/src/agents/tracing/processors.py +++ b/src/agents/tracing/processors.py @@ -739,23 +739,45 @@ def _export_batches(self, deadline: float | None = None): _global_lock = threading.Lock() -def _discard_shut_down_defaults() -> None: - """Drop the cached defaults once shutdown has made them terminal. +def _replacement_exporter(exporter: BackendSpanExporter) -> BackendSpanExporter: + """A live exporter configured exactly like the shut-down one it replaces. - Shutdown is terminal, and the processor owns the exporter it was handed, so the pair - goes down together. Handing the shut-down exporter to a new processor would look like - it worked while silently abandoning every retry backoff -- the first 5xx drops the - batch instead of retrying -- so a later tracing initialization gets a fresh pair. + 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. - Callers must hold ``_global_lock``. + 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) or ( - _global_processor is not None and _global_processor.is_shut_down - ): - _global_exporter = None + 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 @@ -764,7 +786,7 @@ def default_exporter() -> BackendSpanExporter: global _global_exporter with _global_lock: - _discard_shut_down_defaults() + _replace_shut_down_defaults() if _global_exporter is None: _global_exporter = BackendSpanExporter() return _global_exporter @@ -776,7 +798,7 @@ def default_processor() -> BatchTraceProcessor: global _global_processor with _global_lock: - _discard_shut_down_defaults() + _replace_shut_down_defaults() if _global_processor is None: if _global_exporter is None: _global_exporter = BackendSpanExporter() 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 f71c9e5a71..c001d7719f 100644 --- a/src/agents/tracing/setup.py +++ b/src/agents/tracing/setup.py @@ -21,19 +21,41 @@ _ATEXIT_SHUTDOWN_STARTED = False -def _default_stack_is_terminal() -> bool: - """Whether the bootstrapped provider's default processor has been shut down. +def _default_processor_is_terminal() -> bool: + """Whether the default processor we bootstrapped has been shut down. - Shutdown is terminal for the whole default stack -- provider, processor and exporter - alike -- so the next use re-initializes 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. + 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 @@ -73,12 +95,12 @@ def get_trace_provider() -> TraceProvider: global _DEFAULT_PROCESSOR provider = GLOBAL_TRACE_PROVIDER - if provider is not None and not _default_stack_is_terminal(): + if provider is not None and not _default_processor_is_terminal(): return provider with _GLOBAL_TRACE_PROVIDER_LOCK: provider = GLOBAL_TRACE_PROVIDER - if provider is None or _default_stack_is_terminal(): + if provider is None: from .processors import default_processor from .provider import DefaultTraceProvider @@ -87,6 +109,8 @@ def get_trace_provider() -> TraceProvider: 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 a5c8b2ce55..864b052142 100644 --- a/tests/test_trace_processor.py +++ b/tests/test_trace_processor.py @@ -25,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.""" @@ -1439,8 +1441,10 @@ def test_default_pair_is_replaced_after_shutdown(restore_tracing_defaults): assert not second_processor.is_shut_down -def test_default_pair_is_replaced_when_only_the_processor_shut_down(restore_tracing_defaults): - """A no-timeout shutdown never signals the exporter, but the pair is still terminal.""" +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() @@ -1448,10 +1452,10 @@ def test_default_pair_is_replaced_when_only_the_processor_shut_down(restore_trac first_processor.shutdown() assert not first_exporter.is_shut_down - first_exporter.close() - assert tracing_processors.default_exporter() is not first_exporter + 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): @@ -1513,31 +1517,73 @@ def test_fresh_default_exporter_still_retries_transient_failures(restore_tracing assert mock_client.return_value.post.call_count == 3 # max_retries counts attempts -def test_shut_down_provider_reinitializes_with_the_fresh_pair(restore_tracing_defaults): - """A stale provider must not keep exporting through the shut-down pair.""" +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 - first_provider = tracing_setup.get_trace_provider() - first_processor = tracing_processors.default_processor() + provider = tracing_setup.get_trace_provider() first_exporter = tracing_processors.default_exporter() - assert tracing_setup._DEFAULT_PROCESSOR is first_processor + 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) - first_provider.shutdown(timeout=1.0) + provider.shutdown(timeout=1.0) first_exporter.close() - # Reconfiguring must land on the exporter the live provider actually exports through. - set_tracing_export_api_key("fresh_key") + 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 - second_provider = tracing_setup.get_trace_provider() - assert second_provider is not first_provider + tracing_setup.get_trace_provider() + fresh = tracing_processors.default_processor() - registered = cast(Any, second_provider)._multi_processor._processors - assert len(registered) == 1 - assert registered[0] is tracing_processors.default_processor() - assert registered[0] is not first_processor - assert registered[0]._exporter is tracing_processors.default_exporter() - assert registered[0]._exporter is not first_exporter - assert registered[0]._exporter.api_key == "fresh_key" + 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): @@ -1546,7 +1592,7 @@ def test_live_provider_is_not_replaced(restore_tracing_defaults): assert tracing_setup.get_trace_provider() is provider -def test_caller_supplied_provider_is_never_replaced(restore_tracing_defaults): +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")) @@ -1557,6 +1603,26 @@ def test_caller_supplied_provider_is_never_replaced(restore_tracing_defaults): 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]) + + 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): From d8743a787cbf200d2f82df31ba9c3d21cf3b7111 Mon Sep 17 00:00:00 2001 From: rajarshidattapy Date: Fri, 28 Aug 2026 15:43:46 +0530 Subject: [PATCH 4/4] fix(tracing): cast provider for timeout-aware shutdown in tests get_trace_provider() is typed as the base TraceProvider, whose shutdown() takes no timeout, so pyright rejected the timeout kwarg. Cast at the two call sites, matching how the surrounding assertions already reach into the concrete provider. --- tests/test_trace_processor.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_trace_processor.py b/tests/test_trace_processor.py index 864b052142..ca506d8380 100644 --- a/tests/test_trace_processor.py +++ b/tests/test_trace_processor.py @@ -1536,7 +1536,7 @@ def test_recovery_preserves_everything_configured_on_the_stack(restore_tracing_d mine = SpanProcessorForTests() provider.register_processor(mine) - provider.shutdown(timeout=1.0) + cast(Any, provider).shutdown(timeout=1.0) first_exporter.close() recovered = tracing_setup.get_trace_provider() @@ -1615,7 +1615,7 @@ def test_set_trace_processors_opt_out_survives_recovery(restore_tracing_defaults mine = SpanProcessorForTests() provider.set_processors([mine]) - provider.shutdown(timeout=1.0) + cast(Any, provider).shutdown(timeout=1.0) tracing_processors.default_exporter().close() recovered = tracing_setup.get_trace_provider()