fix(tracing): skip obs wrapper inside Temporal activities

start_span/end_span run as SEPARATE Temporal activities (START_SPAN/END_SPAN)
that Temporal can route to different worker processes. The obs wrapper handle
is stored in a process-local module dict, so on a multi-replica fleet the END
lands on a different worker than the START: the handle is never popped (leak /
OOM risk) and the wrapper span is never ended (dangling obs_span_id in Tempo).

Inside a Temporal activity, skip opening our own wrapper and instead stamp the
reverse tag onto the interceptor-propagated ambient span (tag_ambient_obs_span)
and read forward ids via obs_correlation(). Trace-level correlation is preserved
via the Temporal OTel TracingInterceptor (#485); the per-step named wrapper and
TurnTrace RETRY/ASYNC roll-up are deferred (see TODO(obs-followup)).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: NiteshDhanpal

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

@@ -31,7 +31,7 @@
 
 from agentex.lib.core.tracing.obs_ids import LGTM, get_obs_mode
 
-__all__ = ("ObsSpanHandle", "open_obs_span", "close_obs_span")
+__all__ = ("ObsSpanHandle", "open_obs_span", "close_obs_span", "tag_ambient_obs_span")
 
 # Instrumentation scope name so these wrapper spans are identifiable in Tempo/DD.
 _TRACER_NAME = "agentex.business"
@@ -175,6 +175,43 @@ def open_obs_span(
         return None
 
 
+def tag_ambient_obs_span(
+    business_span_id: Optional[str] = None,
+    business_trace_id: Optional[str] = None,
+) -> None:
+    """Stamp the reverse tag onto the CURRENTLY ACTIVE obs span -- without opening
+    a new one.
+
+    Used on the Temporal path (see ``trace._in_temporal_activity``): there we must
+    NOT open our own wrapper span, because start_span/end_span run as separate
+    activities on possibly different workers and the wrapper could never be
+    closed. Instead we lean on the span the Temporal OTel ``TracingInterceptor``
+    already made active for this activity and just add
+    ``agentex.business_span_id`` / ``agentex.business_trace_id`` so the obs -> business
+    pivot still works. Best-effort; never raises."""
+    try:
+        if get_obs_mode() == LGTM:
+            from opentelemetry import trace
+
+            span = trace.get_current_span()
+            if span is not None and span.get_span_context().is_valid:
+                if business_span_id:
+                    span.set_attribute(_ATTR_BUSINESS_SPAN_ID, business_span_id)
+                if business_trace_id:
+                    span.set_attribute(_ATTR_BUSINESS_TRACE_ID, business_trace_id)
+        else:
+            from ddtrace.trace import tracer
+
+            span = tracer.current_span()
+            if span is not None:
+                if business_span_id:
+                    span.set_tag(_ATTR_BUSINESS_SPAN_ID, business_span_id)
+                if business_trace_id:
+                    span.set_tag(_ATTR_BUSINESS_TRACE_ID, business_trace_id)
+    except Exception:  # pragma: no cover - best-effort; obs must never break a call
+        pass
+
+
 def close_obs_span(
     handle: Optional[ObsSpanHandle],
     error: Optional[Dict[str, str]] = None,

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

@@ -16,6 +16,7 @@
     ObsSpanHandle,
     open_obs_span,
     close_obs_span,
+    tag_ambient_obs_span,
 )
 from agentex.lib.core.tracing.span_error import get_span_error, set_span_error
 from agentex.lib.core.tracing.span_queue import (
@@ -42,6 +43,43 @@
 _OBS_HANDLES: dict[str, ObsSpanHandle] = {}
 
 
+def _in_temporal_activity() -> bool:
+    """True when executing inside a Temporal activity.
+
+    On the Temporal path ``start_span`` and ``end_span`` run as SEPARATE
+    activities (START_SPAN / END_SPAN) that Temporal can route to DIFFERENT
+    worker processes. A wrapper obs span opened in the START_SPAN activity could
+    therefore never be closed by END_SPAN -- its handle lives in another
+    process's ``_OBS_HANDLES`` -- so it would leak (unbounded, OOM risk) and its
+    persisted ``obs_span_id`` would dangle (the span is never .end()ed, so never
+    exported to Tempo).
+
+    So inside an activity we do NOT open our own wrapper. We lean on the span the
+    Temporal OTel ``TracingInterceptor`` (see ``core/tracing/temporal.py`` +
+    scale-agentex-python#485) already made active for this activity -- which is
+    rooted under the turn's propagated trace -- and merely stamp the reverse tag
+    onto it (``tag_ambient_obs_span``). That keeps trace-level correlation with
+    no cross-process handle to leak.
+
+    Never raises; returns False when temporalio isn't importable.
+
+    TODO(obs-followup): this intentionally drops the *named per-step* wrapper on
+    the Temporal path (obs_span_id becomes the ambient activity span, not a
+    step-named span) and does NOT add TurnTrace RETRY/ASYNC roll-up -- retried
+    turns still surface as N unlinked spans. Follow-up diff should (a) optionally
+    materialize a self-contained named wrapper inside a single activity using the
+    span's own start/end timestamps, and (b) build the TurnTrace roll-up.
+    Test-later: on a multi-replica worker fleet, assert _OBS_HANDLES stays
+    bounded (no leak / OOM) and that obs_trace_id resolves to the turn trace.
+    """
+    try:
+        from temporalio import activity
+
+        return activity.in_activity()
+    except Exception:
+        return False
+
+
 class Trace:
     """
     Trace is a wrapper around the Agentex API for tracing.
@@ -105,9 +143,20 @@ def start_span(
         # so you can pivot obs -> business in Tempo/DD. Falls back to the ambient
         # obs context (ddtrace) when not in lgtm mode. Business trace_id stays the
         # run-level task id.
+        #
+        # Inside a Temporal activity we skip the wrapper entirely and only tag the
+        # interceptor-propagated ambient span: opening a wrapper there would leak,
+        # since start_span / end_span run as separate activities on possibly
+        # different workers and the handle could never be closed. See
+        # _in_temporal_activity().
         id = str(uuid.uuid4())
-        obs_handle = open_obs_span(name, business_span_id=id, business_trace_id=self.trace_id)
-        obs = obs_handle.correlation if obs_handle is not None else obs_correlation()
+        if _in_temporal_activity():
+            tag_ambient_obs_span(business_span_id=id, business_trace_id=self.trace_id)
+            obs_handle = None
+            obs = obs_correlation()
+        else:
+            obs_handle = open_obs_span(name, business_span_id=id, business_trace_id=self.trace_id)
+            obs = obs_handle.correlation if obs_handle is not None else obs_correlation()
         if obs:
             serialized_data = {**(serialized_data or {}), **obs}
 
@@ -274,9 +323,20 @@ async def start_span(
         # so you can pivot obs -> business in Tempo/DD. Falls back to the ambient
         # obs context (ddtrace) when not in lgtm mode. Business trace_id stays the
         # run-level task id.
+        #
+        # Inside a Temporal activity we skip the wrapper entirely and only tag the
+        # interceptor-propagated ambient span: opening a wrapper there would leak,
+        # since start_span / end_span run as separate activities on possibly
+        # different workers and the handle could never be closed. See
+        # _in_temporal_activity().
         id = str(uuid.uuid4())
-        obs_handle = open_obs_span(name, business_span_id=id, business_trace_id=self.trace_id)
-        obs = obs_handle.correlation if obs_handle is not None else obs_correlation()
+        if _in_temporal_activity():
+            tag_ambient_obs_span(business_span_id=id, business_trace_id=self.trace_id)
+            obs_handle = None
+            obs = obs_correlation()
+        else:
+            obs_handle = open_obs_span(name, business_span_id=id, business_trace_id=self.trace_id)
+            obs = obs_handle.correlation if obs_handle is not None else obs_correlation()
         if obs:
             serialized_data = {**(serialized_data or {}), **obs}