fix(sdk): support OpenTelemetry 1.39+ OTLPSpanExporter and prevent 401 race on logfile upload

agentops/client/client.py

@@ -25,6 +25,8 @@ def _end_init_trace_atexit():
     if _client_init_trace_context is not None:
         logger.debug("Auto-ending client's init trace during shutdown.")
         try:
+            if Client._Client__instance is not None:
+                Client._Client__instance.wait_for_auth()
             # Use global tracer to end the trace directly
             if tracer.initialized and _client_init_trace_context.span.is_recording():
                 tracer.end_trace(_client_init_trace_context, end_state="Shutdown")
@@ -77,8 +79,16 @@ def __init__(self):
             # self._init_trace_context = None # Already done in __new__
             # self._legacy_session_for_init_trace = None # Already done in __new__
 
+    def wait_for_auth(self, timeout: float = 10.0) -> None:
+        """Wait for any pending background authentication thread to complete."""
+        auth_thread = getattr(self, "_auth_thread", None)
+        if auth_thread is not None and auth_thread.is_alive():
+            auth_thread.join(timeout=timeout)
+
     def get_current_jwt(self) -> Optional[str]:
         """Get the current JWT token."""
+        if self._auth_token is None:
+            self.wait_for_auth()
         with self._auth_lock:
             return self._auth_token
 
@@ -114,6 +124,12 @@ async def _fetch_auth_async(self, api_key: str) -> Optional[dict]:
                 return None
         except Exception:
             return None
+        finally:
+            try:
+                from agentops.client.http.http_client import HttpClient
+                await HttpClient.close_async_session()
+            except Exception:
+                pass
 
     def _start_auth_task(self, api_key: str):
         """Start the async authentication task."""
@@ -133,7 +149,9 @@ def run_async_auth():
                 import threading
 
                 auth_thread = threading.Thread(target=run_async_auth, daemon=True)
+                self._auth_thread = auth_thread
                 auth_thread.start()
+                self.wait_for_auth(timeout=15.0)
         except RuntimeError:
             # Create new event loop in background thread
             def run_async_auth():
@@ -142,7 +160,9 @@ def run_async_auth():
             import threading
 
             auth_thread = threading.Thread(target=run_async_auth, daemon=True)
+            self._auth_thread = auth_thread
             auth_thread.start()
+            self.wait_for_auth(timeout=15.0)
 
     def init(self, **kwargs: Any) -> None:  # Return type updated to None
         # Recreate the Config object to parse environment variables at the time of initialization

agentops/logging/instrument_logging.py

@@ -88,7 +88,10 @@ def upload_logfile(trace_id: int) -> None:
         return
 
     client = get_client()
-    client.api.v4.upload_logfile(log_content, trace_id)
+    if hasattr(client, "wait_for_auth"):
+        client.wait_for_auth()
+    if getattr(client.api.v4, "auth_token", None):
+        client.api.v4.upload_logfile(log_content, trace_id)
 
     # Clear the buffer after upload
     _log_buffer.seek(0)

agentops/sdk/exporters.py

@@ -67,6 +67,11 @@ def __init__(
             parent_kwargs["compression"] = compression
 
         super().__init__(endpoint=endpoint, **parent_kwargs)
+        if not hasattr(self, "_headers"):
+            self._headers = getattr(getattr(self, "_client", None), "_headers", None) or (filtered_headers or {})
+        if not hasattr(self, "_session"):
+            transport = getattr(getattr(self, "_client", None), "_transport", None)
+            self._session = getattr(transport, "_session", None) or requests.Session()
 
     def _get_current_jwt(self) -> Optional[str]:
         """Get the current JWT token from the provider or stored JWT."""
@@ -139,6 +144,8 @@ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
             # Temporarily update the session headers for this request
             original_headers = dict(self._session.headers)
             self._session.headers.update(current_headers)
+            if hasattr(self, "_client") and hasattr(self._client, "_headers"):
+                self._client._headers.update(current_headers)
 
             try:
                 # Call parent export method