fix(tracing): rebuild the default exporter and processor after shutdown

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,40 +739,68 @@ 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:
     """The default processor, which exports traces and spans to the backend in batches."""
     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

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.

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)