fix(tracing): keep extra span keys, make the processor lock reentrant, drop the tracer's dead client

Review follow-ups on the span-processor removal:
- Span keeps unknown keys like the generated model it replaces, so a
  custom processor's own attributes and older worker payloads survive.
- set_processor_configs re-enters add_processor_config under the manager
  lock, which deadlocked on a plain Lock. Reentrant now, with tests for the
  SGP happy path and the batch registration.
- Trace and Tracer no longer use the client they receive, so it is optional
  and the ADK tracing module stops building an httpx client per event loop.
- Changelog entry for the breaking removals, stale prose updated.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Author: mohammadatallah-scale

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 ### ⚠ BREAKING CHANGES
 
+* **tracing:** removed the Agentex-native span processor and its `AgentexTracingProcessorConfig` (the Agentex server is retiring its Postgres spans API), along with `Trace.get_span` / `Trace.list_spans` and their async twins. `SGPTracingProcessorConfig` is the only processor config and registering any other type raises `ValueError`. The in-memory `Span` is now `agentex.lib.types.tracing.Span` (also exported as `agentex.lib.core.tracing.Span`); the generated `agentex.types.span.Span` disappears with the next client generation. Its `to_dict()` / `to_json()` return the full JSON-mode dump rather than only the fields that were set.
 * **harness:** removed the deprecated bespoke LangGraph tracing handler `create_langgraph_tracing_handler` (and its `AgentexLangGraphTracingHandler` class) from the public `agentex.lib.adk` surface. Span tracing is now derived from the canonical `StreamTaskMessage*` stream by `UnifiedEmitter` — wrap your run in the harness `*Turn` and drive `UnifiedEmitter.yield_turn` / `auto_send_turn`. The `agentex init` templates were migrated accordingly.
 * **harness:** removed the deprecated bespoke Pydantic-AI tracing handler `create_pydantic_ai_tracing_handler` (and its `AgentexPydanticAITracingHandler` class) from the public `agentex.lib.adk` surface. Span tracing is now derived from the canonical `StreamTaskMessage*` stream by `UnifiedEmitter` — wrap your run in `PydanticAITurn` and drive `UnifiedEmitter.yield_turn` / `auto_send_turn`. The `agentex init` templates were migrated accordingly.
 * **harness:** each harness now exposes exactly `_<harness>_sync.py` + `_<harness>_turn.py` under `agentex.lib.adk._modules`. The OpenAI harness `OpenAITurn` and `convert_openai_to_agentex_events` moved to `agentex.lib.adk._modules._openai_turn` / `_openai_sync`; back-compat shims remain at `agentex.lib.adk.providers._modules.{openai_turn,sync_provider}` for one release. Public facade names (`stream_pydantic_ai_events`, `stream_langgraph_events`, `emit_langgraph_messages`, etc.) are unchanged.

src/agentex/lib/adk/_modules/tracing.py

@@ -11,7 +11,6 @@
 from temporalio.exceptions import ActivityError, TimeoutError as TemporalTimeoutError, is_cancelled_exception
 
 from agentex import AsyncAgentex  # noqa: F401
-from agentex.lib.adk.utils._modules.client import create_async_agentex_client
 from agentex.lib.core.services.adk.tracing import TracingService
 from agentex.lib.core.temporal.activities.activity_helpers import ActivityHelpers
 from agentex.lib.core.temporal.activities.adk.tracing_activities import (
@@ -145,46 +144,17 @@ def __init__(self, tracing_service: TracingService | None = None):
 
         Args:
             tracing_service (Optional[TracingService]): Optional pre-configured tracing service.
-                If None, will be lazily created on first use so the httpx client is
-                bound to the correct running event loop.
+                If None, one is created on first use.
         """
         self._tracing_service_explicit = tracing_service
         self._tracing_service_lazy: TracingService | None = None
-        self._bound_loop_id: int | None = None
 
     @property
     def _tracing_service(self) -> TracingService:
         if self._tracing_service_explicit is not None:
             return self._tracing_service_explicit
-
-        import asyncio
-
-        # Determine the current event loop (if any).
-        try:
-            loop = asyncio.get_running_loop()
-            loop_id = id(loop)
-        except RuntimeError:
-            loop_id = None
-
-        # Re-create the underlying httpx client when the event loop changes
-        # (e.g. between HTTP requests in a sync ASGI server) to avoid
-        # "Event loop is closed" / "bound to a different event loop" errors.
-        if self._tracing_service_lazy is None or (loop_id is not None and loop_id != self._bound_loop_id):
-            import httpx
-
-            # Keepalive ON: connections are reused within a single event
-            # loop, eliminating the TLS-handshake-per-span penalty under
-            # load.  Cross-loop safety is preserved by rebuilding the
-            # client whenever loop_id changes (the conditional above).
-            agentex_client = create_async_agentex_client(
-                http_client=httpx.AsyncClient(
-                    limits=httpx.Limits(max_keepalive_connections=20),
-                ),
-            )
-            tracer = AsyncTracer(agentex_client)
-            self._tracing_service_lazy = TracingService(tracer=tracer)
-            self._bound_loop_id = loop_id
-
+        if self._tracing_service_lazy is None:
+            self._tracing_service_lazy = TracingService(tracer=AsyncTracer())
         return self._tracing_service_lazy
 
     @asynccontextmanager

src/agentex/lib/core/tracing/processors/sgp_tracing_processor.py

@@ -74,8 +74,8 @@ def _sgp_metadata(span: Span) -> Any:
 
     Returns a COPY rather than mutating ``span``. ``trace.py`` hands the same
     Span instance to every registered processor, so anything written onto
-    ``span.data`` here would also be serialized by the Agentex processor and
-    show up in caller-visible span data. ``__commit_sha__`` is opt-in and
+    ``span.data`` here would also reach every other processor and show up in
+    caller-visible span data. ``__commit_sha__`` is opt-in and
     SGP-scoped, so it must not leak that way.
 
     (The ``__source__`` / ``__agent_*`` keys set by ``_add_source_to_span`` do

src/agentex/lib/core/tracing/span_error.py

@@ -12,12 +12,9 @@
 from agentex.lib.types.tracing import Span
 
 # Reserved key under ``Span.data`` carrying failure info for a span whose
-# context-manager body raised. Mirrors the existing ``__span_type__`` /
-# ``__source__`` reserved-key convention already read/written by the SGP
-# processor. Stored in ``data`` because the Span model is generated from the
-# OpenAPI spec and has no first-class status/error field; ``data`` is a real
-# field, so it survives ``model_copy(deep=True)`` and round-trips to both the
-# SGP and agentex-native span stores.
+# context-manager body raised, alongside the ``__span_type__`` / ``__source__``
+# keys the SGP processor already reads. Kept in ``data`` so it survives
+# ``model_copy(deep=True)`` and reaches the processors with the span.
 SPAN_ERROR_KEY = "__error__"
 
 ERROR_CATEGORY_UNKNOWN: ErrorCategory = "unknown"

src/agentex/lib/core/tracing/trace.py

@@ -218,22 +218,24 @@ def _begin_obs(
 
 class Trace:
     """
-    Trace is a wrapper around the Agentex API for tracing.
-    It provides a context manager for spans and a way to start and end spans.
+    Trace groups the spans of one trace id and hands each span to the
+    registered processors. It provides a context manager for spans and a way
+    to start and end spans.
     """
 
     def __init__(
         self,
         processors: list[SyncTracingProcessor],
-        client: Agentex,
+        client: Agentex | None = None,
         trace_id: str | None = None,
     ):
         """
         Initialize a new trace with the specified trace ID.
 
         Args:
-            trace_id: Required trace ID to use for this trace.
-            processors: Optional list of tracing processors to use for this trace.
+            processors: Tracing processors every span is handed to.
+            client: Kept for backward compatibility, no longer used.
+            trace_id: Trace ID to use for this trace.
         """
         self.processors = processors
         self.client = client
@@ -251,7 +253,7 @@ def start_span(
         task_id: str | None = None,
     ) -> Span:
         """
-        Start a new span and register it with the API.
+        Start a new span and hand it to the registered processors.
 
         Args:
             name: Name of the span.
@@ -325,8 +327,6 @@ def end_span(
 
         return span
 
-
-
     @contextmanager
     def span(
         self,
@@ -355,14 +355,14 @@ def span(
 
 class AsyncTrace:
     """
-    AsyncTrace is a wrapper around the Agentex API for tracing.
-    It provides a context manager for spans and a way to start and end spans.
+    AsyncTrace is the async version of Trace. It provides a context manager
+    for spans and a way to start and end spans.
     """
 
     def __init__(
         self,
         processors: list[AsyncTracingProcessor],
-        client: AsyncAgentex,
+        client: AsyncAgentex | None = None,
         trace_id: str | None = None,
         span_queue: AsyncSpanQueue | None = None,
     ):
@@ -391,7 +391,7 @@ async def start_span(
         task_id: str | None = None,
     ) -> Span:
         """
-        Start a new span and register it with the API.
+        Start a new span and hand it to the registered processors.
 
         Args:
             name: Name of the span.
@@ -480,8 +480,6 @@ async def end_span(
 
         return span
 
-
-
     @asynccontextmanager
     async def span(
         self,

src/agentex/lib/core/tracing/tracer.py

@@ -15,12 +15,12 @@ class Tracer:
     It manages the client connection and creates traces.
     """
 
-    def __init__(self, client: Agentex):
+    def __init__(self, client: Agentex | None = None):
         """
-        Initialize a new sync tracer with the provided client.
+        Initialize a new sync tracer.
 
         Args:
-            client: Agentex client instance used for API communication.
+            client: Kept for backward compatibility, no longer used.
         """
         self.client = client
 
@@ -47,12 +47,12 @@ class AsyncTracer:
     It manages the async client connection and creates async traces.
     """
 
-    def __init__(self, client: AsyncAgentex):
+    def __init__(self, client: AsyncAgentex | None = None):
         """
-        Initialize a new async tracer with the provided client.
+        Initialize a new async tracer.
 
         Args:
-            client: AsyncAgentex client instance used for API communication.
+            client: Kept for backward compatibility, no longer used.
         """
         self.client = client
 

src/agentex/lib/core/tracing/tracing_processor_manager.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from threading import Lock
+from threading import RLock
 
 from agentex.lib.types.tracing import TracingProcessorConfig
 from agentex.lib.core.tracing.processors.sgp_tracing_processor import (
@@ -24,7 +24,8 @@ def __init__(self):
         # Cache for processors
         self.sync_processors: list[SyncTracingProcessor] = []
         self.async_processors: list[AsyncTracingProcessor] = []
-        self.lock = Lock()
+        # Reentrant: set_processor_configs holds it while calling add_processor_config.
+        self.lock = RLock()
 
     def add_processor_config(self, processor_config: TracingProcessorConfig) -> None:
         with self.lock:

src/agentex/lib/types/tracing.py

@@ -3,6 +3,8 @@
 from typing import Any, Literal
 from datetime import datetime
 
+from pydantic import ConfigDict
+
 from agentex.lib.utils.model_utils import BaseModel
 
 
@@ -22,6 +24,9 @@ class BaseModelWithTraceParams(BaseModel):
 class Span(BaseModel):
     """In-memory span handed to tracing processors. Owned here, not by the generated client."""
 
+    # The generated model kept unknown keys, and custom processors may stash their own.
+    model_config = ConfigDict(extra="allow")
+
     id: str
     name: str
     start_time: datetime

tests/lib/core/tracing/test_span_model.py

@@ -0,0 +1,21 @@
+from datetime import UTC, datetime
+
+from agentex.lib.types.tracing import Span
+
+
+def _span(**extra) -> Span:
+    return Span(id="s1", name="n", trace_id="t1", start_time=datetime(2026, 1, 1, tzinfo=UTC), **extra)
+
+
+def test_unknown_keys_survive_validation_and_a_json_round_trip():
+    span = Span.model_validate({**_span().model_dump(), "extension": {"sampled": True}})
+
+    assert span.extension == {"sampled": True}  # type: ignore[attr-defined]
+    assert Span.model_validate_json(span.model_dump_json()).model_dump()["extension"] == {"sampled": True}
+
+
+def test_processors_can_attach_their_own_attributes():
+    span = _span()
+    span.annotation = "custom"  # type: ignore[attr-defined]
+
+    assert span.model_copy(deep=True).model_dump()["annotation"] == "custom"

tests/lib/core/tracing/test_tracing_processor_manager.py

@@ -1,8 +1,32 @@
+import threading
 from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
 
 import pytest
 
+from agentex.lib.types.tracing import SGPTracingProcessorConfig
 from agentex.lib.core.tracing.tracing_processor_manager import TracingProcessorManager
+from agentex.lib.core.tracing.processors.sgp_tracing_processor import (
+    SGPSyncTracingProcessor,
+    SGPAsyncTracingProcessor,
+)
+
+SGP_MODULE = "agentex.lib.core.tracing.processors.sgp_tracing_processor"
+
+
+def _sgp_config() -> SGPTracingProcessorConfig:
+    return SGPTracingProcessorConfig(sgp_api_key="k", sgp_account_id="a", sgp_base_url="http://sgp.test")
+
+
+def _patched_sgp():
+    env = MagicMock()
+    env.refresh.return_value = MagicMock(ACP_TYPE=None, AGENT_NAME=None, AGENT_ID=None, AGENT_VERSION=None)
+    return (
+        patch(f"{SGP_MODULE}.SGPClient"),
+        patch(f"{SGP_MODULE}.AsyncSGPClient"),
+        patch(f"{SGP_MODULE}.tracing.init"),
+        patch(f"{SGP_MODULE}.EnvironmentVariables", env),
+    )
 
 
 def test_unknown_processor_type_is_rejected_by_name():
@@ -13,3 +37,32 @@ def test_unknown_processor_type_is_rejected_by_name():
 
     assert manager.get_sync_processors() == []
     assert manager.get_async_processors() == []
+
+
+def test_sgp_config_registers_one_sync_and_one_async_processor():
+    p1, p2, p3, p4 = _patched_sgp()
+    with p1, p2, p3, p4:
+        manager = TracingProcessorManager()
+        manager.add_processor_config(_sgp_config())
+
+    (sync_processor,) = manager.get_sync_processors()
+    (async_processor,) = manager.get_async_processors()
+    assert isinstance(sync_processor, SGPSyncTracingProcessor)
+    assert isinstance(async_processor, SGPAsyncTracingProcessor)
+
+
+def test_set_processor_configs_registers_every_config_without_deadlocking():
+    p1, p2, p3, p4 = _patched_sgp()
+    manager = TracingProcessorManager()
+    done = threading.Event()
+
+    def register():
+        with p1, p2, p3, p4:
+            manager.set_processor_configs([_sgp_config(), _sgp_config()])
+        done.set()
+
+    threading.Thread(target=register, daemon=True).start()
+
+    assert done.wait(timeout=5), "set_processor_configs hung: the manager lock must be reentrant"
+    assert len(manager.get_sync_processors()) == 2
+    assert len(manager.get_async_processors()) == 2