From c092c1722db3ec09a46ac3021fb0ada6e22120a4 Mon Sep 17 00:00:00 2001 From: Nitesh Dhanpal Date: Sat, 1 Aug 2026 10:58:30 -0700 Subject: [PATCH 1/4] feat(tracing): correlate business spans with obs via dedicated wrapper span Model the business<->observability correlation edge on the emit side, with no schema migration (rides the existing operation_metadata JSONB). - obs_ids: standardize the correlation keys to obs_trace_id/obs_span_id (underscored, JSON-path friendly); remove the non-working `dual` mode (it required an in-process ddtrace<->OTel bridge that can't exist -- you can't run ddtrace-run and the OTel operator together, and DD_TRACE_OTEL_ENABLED is a single tracer). `dual` now safely degrades to dd_only. Harden obs_correlation to never raise. - obs_span (new): when the SDK creates a business span it opens a dedicated obs span named for that step and makes it active, so obs_span_id is stable and meaningful (a named span with its httpx call nested underneath) instead of an arbitrary innermost instrumentation span. Backends: OTel in lgtm; ddtrace in dd_only but only when a request trace is already active (avoids orphan root traces in un-instrumented agents). Reverse tag: stamps agentex.business_span_id / agentex.business_trace_id onto the obs span so the pivot is bidirectional. - trace: wire the wrapper into start_span/end_span (sync + async). Observability can never fail an app call -- every path is guarded and is a no-op when the tracer isn't configured. - tests: obs_ids (mode degrade + keys), obs_span (both backends, non-interference, never-fails, reverse tag), and the 3-turn mortgage Turn-2 example pinned as an executable contract. Co-Authored-By: Claude Opus 4.8 --- src/agentex/lib/core/tracing/obs_ids.py | 33 +- src/agentex/lib/core/tracing/obs_span.py | 153 +++++++++ src/agentex/lib/core/tracing/trace.py | 51 ++- tests/lib/core/tracing/test_obs_ids.py | 127 ++++++++ tests/lib/core/tracing/test_obs_span.py | 399 +++++++++++++++++++++++ 5 files changed, 740 insertions(+), 23 deletions(-) create mode 100644 src/agentex/lib/core/tracing/obs_span.py create mode 100644 tests/lib/core/tracing/test_obs_ids.py create mode 100644 tests/lib/core/tracing/test_obs_span.py diff --git a/src/agentex/lib/core/tracing/obs_ids.py b/src/agentex/lib/core/tracing/obs_ids.py index 99c6b2555..bac500b20 100644 --- a/src/agentex/lib/core/tracing/obs_ids.py +++ b/src/agentex/lib/core/tracing/obs_ids.py @@ -11,11 +11,16 @@ persisted business span to the Tempo/Datadog trace for the turn that produced it, while the business trace still groups the entire run by task id. -Source selection follows SGP_OBS_MODE, matching egp-api-backend: +Source selection follows SGP_OBS_MODE: - unset / "dd_only": ddtrace context (current stack) - - "dual": OTel/LGTM preferred, ddtrace fallback - "lgtm": OTel/LGTM only +("dual" was removed: co-resident ddtrace+OTel can't be bridged in-process -- +you can't run ddtrace-run and the OTel operator's auto-instrumentation in the +same process, and DD_TRACE_OTEL_ENABLED yields a single tracer with nothing to +bridge. Two-backend export is a collector fan-out under "lgtm", not a mode here. +An unrecognized SGP_OBS_MODE -- including a stale "dual" -- degrades to dd_only.) + This never fabricates ids -- if no observability context is active, it returns an empty dict and the span is simply not tagged. """ @@ -27,10 +32,9 @@ __all__ = ("get_obs_mode", "obs_correlation") DD_ONLY = "dd_only" -DUAL = "dual" LGTM = "lgtm" _DEFAULT_MODE = DD_ONLY -_VALID_MODES = (DD_ONLY, DUAL, LGTM) +_VALID_MODES = (DD_ONLY, LGTM) def get_obs_mode() -> str: @@ -65,19 +69,22 @@ def _ddtrace_ids() -> Optional[Tuple[str, str]]: def obs_correlation() -> Dict[str, str]: - """Return ``{"obs.trace_id": ..., "obs.span_id": ...}`` for the active + """Return ``{"obs_trace_id": ..., "obs_span_id": ...}`` for the active observability context, or ``{}`` if none is active. + These land in the business span's ``data`` -> egp ``operation_metadata`` + (an existing JSONB column, GIN-indexed) -> ClickHouse ``metadata_raw``, so + the correlation edge needs no schema migration. Underscored keys (not + dotted) keep them addressable via Postgres JSON paths + (``operation_metadata->>'obs_trace_id'``). + Never fabricates ids -- this is a correlation tag, not the span's id. """ - mode = get_obs_mode() - if mode == LGTM: - ids = _lgtm_ids() - elif mode == DUAL: - ids = _lgtm_ids() or _ddtrace_ids() - else: # dd_only - ids = _ddtrace_ids() + try: + ids = _lgtm_ids() if get_obs_mode() == LGTM else _ddtrace_ids() + except Exception: # obs must never fail an app call + return {} if not ids: return {} - return {"obs.trace_id": ids[0], "obs.span_id": ids[1]} + return {"obs_trace_id": ids[0], "obs_span_id": ids[1]} diff --git a/src/agentex/lib/core/tracing/obs_span.py b/src/agentex/lib/core/tracing/obs_span.py new file mode 100644 index 000000000..77186d2ad --- /dev/null +++ b/src/agentex/lib/core/tracing/obs_span.py @@ -0,0 +1,153 @@ +"""Dedicated per-business-span observability wrapper span. + +Capturing obs ids from "whatever instrumentation span happens to be innermost +at emit time" is coarse -- it could be an arbitrary httpx-client span, and every +business span in a request would collapse onto the same request/activity span. + +Instead, when the SDK creates a business span we open a **real obs span named +for that step and make it active**. Then: + - ``obs_span_id`` is stable and meaningful (a span named for the business + step, not an arbitrary leaf), and + - any nested instrumentation (httpx, db, ...) parents under it. + +The wrapper's own trace_id/span_id are read directly from its span context, so +the correlation tag is deterministic regardless of what else is on the stack. + +Backend follows ``SGP_OBS_MODE``: + - ``lgtm`` -> an OpenTelemetry span (the convergence target). + - ``dd_only`` -> a ddtrace span, but ONLY when a ddtrace trace is already + active for the request. Opening one unconditionally would emit orphan root + traces in un-instrumented (bare-uvicorn, no ddtrace-run) agents, so when + nothing is active we return ``None`` and the caller keeps its ambient + behavior. + +No-op when the relevant tracer isn't importable. Never raises -- observability +must never break a business span. +""" +from __future__ import annotations + +from typing import Callable, Dict, Optional + +from agentex.lib.core.tracing.obs_ids import LGTM, get_obs_mode + +__all__ = ("ObsSpanHandle", "open_obs_span", "close_obs_span") + +# Instrumentation scope name so these wrapper spans are identifiable in Tempo/DD. +_TRACER_NAME = "agentex.business" + +# Reverse-tag attribute keys: the business span/trace ids stamped onto the obs +# span so you can pivot obs -> business (search these in Tempo/DD). +_ATTR_BUSINESS_SPAN_ID = "agentex.business_span_id" +_ATTR_BUSINESS_TRACE_ID = "agentex.business_trace_id" + + +class ObsSpanHandle: + """Live handle for an open wrapper span: the correlation tag read from it + plus a backend-specific closer (detach/end or finish).""" + + __slots__ = ("correlation", "_close") + + def __init__(self, correlation: Dict[str, str], close: Callable[[], None]): + self.correlation = correlation + self._close = close + + +def _hex_ids(trace_id: int, span_id: int) -> Dict[str, str]: + """W3C-hex form: 32-hex trace, 16-hex span.""" + return { + "obs_trace_id": format(trace_id, "032x"), + "obs_span_id": format(span_id, "016x"), + } + + +def _open_otel_span( + name: str, + business_span_id: Optional[str], + business_trace_id: Optional[str], +) -> Optional[ObsSpanHandle]: + try: + from opentelemetry import context, trace + except ImportError: + return None + try: + span = trace.get_tracer(_TRACER_NAME).start_span(name) + 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) + token = context.attach(trace.set_span_in_context(span)) + sc = span.get_span_context() + correlation = _hex_ids(sc.trace_id, sc.span_id) if (sc and sc.is_valid) else {} + + def _close() -> None: + try: + context.detach(token) + finally: + span.end() + + return ObsSpanHandle(correlation, _close) + except Exception: # pragma: no cover - best-effort; never break the business span + return None + + +def _open_ddtrace_span( + name: str, + business_span_id: Optional[str], + business_trace_id: Optional[str], +) -> Optional[ObsSpanHandle]: + try: + from ddtrace.trace import tracer + except ImportError: + return None + try: + # Only wrap when ddtrace is actually tracing the request; otherwise a + # wrapper would be an orphan root trace in an un-instrumented process. + if tracer.current_trace_context() is None: + return None + span = tracer.start_span(name, activate=True) + 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) + correlation = _hex_ids(span.trace_id, span.span_id) if span.trace_id else {} + return ObsSpanHandle(correlation, span.finish) + except Exception: # pragma: no cover - best-effort + return None + + +def open_obs_span( + name: str, + business_span_id: Optional[str] = None, + business_trace_id: Optional[str] = None, +) -> Optional[ObsSpanHandle]: + """Open an obs span named ``name`` in the active backend, make it the active + span, and return a handle carrying its ``{"obs_trace_id","obs_span_id"}``. + + ``business_span_id`` / ``business_trace_id`` are stamped onto the obs span as + the reverse tag (``agentex.business_span_id`` / ``agentex.business_trace_id``) + so you can pivot obs -> business by searching them in Tempo/DD. + + Returns ``None`` (so the caller falls back to ambient behavior) when the + backend tracer isn't available or, in ``dd_only``, no request trace is + active. + + Never raises: a top-level guard backstops anything the backend helpers + don't (e.g. a broken tracer install raising on import) so observability can + never fail an app call. + """ + try: + if get_obs_mode() == LGTM: + return _open_otel_span(name, business_span_id, business_trace_id) + return _open_ddtrace_span(name, business_span_id, business_trace_id) + except Exception: # pragma: no cover - backstop; obs must never break a call + return None + + +def close_obs_span(handle: Optional[ObsSpanHandle]) -> None: + """Close the wrapper span (detach + end, or finish). Safe on ``None``.""" + if handle is None: + return + try: + handle._close() + except Exception: # pragma: no cover - best-effort + pass diff --git a/src/agentex/lib/core/tracing/trace.py b/src/agentex/lib/core/tracing/trace.py index c3ec91bc3..031e69e9c 100644 --- a/src/agentex/lib/core/tracing/trace.py +++ b/src/agentex/lib/core/tracing/trace.py @@ -12,6 +12,11 @@ from agentex.lib.utils.logging import make_logger from agentex.lib.utils.model_utils import recursive_model_dump from agentex.lib.core.tracing.obs_ids import obs_correlation +from agentex.lib.core.tracing.obs_span import ( + ObsSpanHandle, + close_obs_span, + open_obs_span, +) from agentex.lib.core.tracing.span_error import set_span_error from agentex.lib.core.tracing.span_queue import ( SpanEventType, @@ -49,6 +54,8 @@ def __init__( self.processors = processors self.client = client self.trace_id = trace_id + # Live per-business-span obs wrapper spans, keyed by business span id. + self._obs_handles: dict[str, ObsSpanHandle] = {} def start_span( self, @@ -80,13 +87,19 @@ def start_span( serialized_input = recursive_model_dump(input) if input else None serialized_data = recursive_model_dump(data) if data else None - # Tag the business span with the active observability trace_id/span_id - # (OTel/ddtrace) so it can be correlated to the per-turn obs trace. The - # business trace_id stays the run-level task id -- see obs_ids.py. - obs = obs_correlation() + # Open a dedicated obs wrapper span named for this step and make it + # active, so obs_span_id is stable/meaningful (not an arbitrary innermost + # httpx span). It also carries the reverse tag (business span/trace id) + # 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. + 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 obs: serialized_data = {**(serialized_data or {}), **obs} - id = str(uuid.uuid4()) span = Span( id=id, @@ -98,6 +111,8 @@ def start_span( data=serialized_data, task_id=task_id, ) + if obs_handle is not None: + self._obs_handles[span.id] = obs_handle for processor in self.processors: processor.on_span_start(span) @@ -120,6 +135,9 @@ def end_span( if span.end_time is None: span.end_time = datetime.now(UTC) + # Close the dedicated obs wrapper span (detach context + end it). + close_obs_span(self._obs_handles.pop(span.id, None)) + span.input = recursive_model_dump(span.input) if span.input else None span.output = recursive_model_dump(span.output) if span.output else None span.data = recursive_model_dump(span.data) if span.data else None @@ -206,6 +224,8 @@ def __init__( self.client = client self.trace_id = trace_id self._span_queue = span_queue or get_default_span_queue() + # Live per-business-span obs wrapper spans, keyed by business span id. + self._obs_handles: dict[str, ObsSpanHandle] = {} async def start_span( self, @@ -236,13 +256,19 @@ async def start_span( serialized_input = recursive_model_dump(input) if input else None serialized_data = recursive_model_dump(data) if data else None - # Tag the business span with the active observability trace_id/span_id - # (OTel/ddtrace) so it can be correlated to the per-turn obs trace. The - # business trace_id stays the run-level task id -- see obs_ids.py. - obs = obs_correlation() + # Open a dedicated obs wrapper span named for this step and make it + # active, so obs_span_id is stable/meaningful (not an arbitrary innermost + # httpx span). It also carries the reverse tag (business span/trace id) + # 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. + 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 obs: serialized_data = {**(serialized_data or {}), **obs} - id = str(uuid.uuid4()) span = Span( id=id, @@ -254,6 +280,8 @@ async def start_span( data=serialized_data, task_id=task_id, ) + if obs_handle is not None: + self._obs_handles[span.id] = obs_handle if self.processors: self._span_queue.enqueue(SpanEventType.START, span.model_copy(deep=True), self.processors) @@ -276,6 +304,9 @@ async def end_span( if span.end_time is None: span.end_time = datetime.now(UTC) + # Close the dedicated obs wrapper span (detach context + end it). + close_obs_span(self._obs_handles.pop(span.id, None)) + span.input = recursive_model_dump(span.input) if span.input else None span.output = recursive_model_dump(span.output) if span.output else None span.data = recursive_model_dump(span.data) if span.data else None diff --git a/tests/lib/core/tracing/test_obs_ids.py b/tests/lib/core/tracing/test_obs_ids.py new file mode 100644 index 000000000..ddd079743 --- /dev/null +++ b/tests/lib/core/tracing/test_obs_ids.py @@ -0,0 +1,127 @@ +from __future__ import annotations + +import sys +import types + +import pytest + +from agentex.lib.core.tracing import obs_ids +from agentex.lib.core.tracing.obs_ids import get_obs_mode, obs_correlation + + +class TestGetObsMode: + @pytest.mark.parametrize( + "raw, expected", + [ + (None, "dd_only"), # unset + ("", "dd_only"), # empty + ("dd_only", "dd_only"), + ("lgtm", "lgtm"), + ("LGTM", "lgtm"), # case-insensitive + (" lgtm ", "lgtm"), # trimmed + ("dual", "dd_only"), # removed mode -> safe degrade + ("garbage", "dd_only"), # unrecognized -> safe degrade + ], + ) + def test_mode_resolution(self, monkeypatch, raw, expected): + if raw is None: + monkeypatch.delenv("SGP_OBS_MODE", raising=False) + else: + monkeypatch.setenv("SGP_OBS_MODE", raw) + assert get_obs_mode() == expected + + +class TestObsCorrelation: + def test_lgtm_mode_reads_otel_and_emits_underscored_keys(self, monkeypatch): + monkeypatch.setenv("SGP_OBS_MODE", "lgtm") + monkeypatch.setattr(obs_ids, "_lgtm_ids", lambda: ("otel_trace", "otel_span")) + # In lgtm mode ddtrace must NOT be consulted. + monkeypatch.setattr( + obs_ids, "_ddtrace_ids", lambda: pytest.fail("ddtrace read in lgtm mode") + ) + + assert obs_correlation() == { + "obs_trace_id": "otel_trace", + "obs_span_id": "otel_span", + } + + def test_dd_only_mode_reads_ddtrace(self, monkeypatch): + monkeypatch.setenv("SGP_OBS_MODE", "dd_only") + monkeypatch.setattr(obs_ids, "_ddtrace_ids", lambda: ("dd_trace", "dd_span")) + monkeypatch.setattr( + obs_ids, "_lgtm_ids", lambda: pytest.fail("otel read in dd_only mode") + ) + + assert obs_correlation() == { + "obs_trace_id": "dd_trace", + "obs_span_id": "dd_span", + } + + def test_stale_dual_degrades_to_ddtrace(self, monkeypatch): + """A leftover SGP_OBS_MODE=dual must behave as dd_only, not read OTel.""" + monkeypatch.setenv("SGP_OBS_MODE", "dual") + monkeypatch.setattr(obs_ids, "_ddtrace_ids", lambda: ("dd_trace", "dd_span")) + monkeypatch.setattr( + obs_ids, "_lgtm_ids", lambda: pytest.fail("otel read for stale dual mode") + ) + + assert obs_correlation() == { + "obs_trace_id": "dd_trace", + "obs_span_id": "dd_span", + } + + def test_no_active_context_returns_empty(self, monkeypatch): + monkeypatch.delenv("SGP_OBS_MODE", raising=False) # dd_only + monkeypatch.setattr(obs_ids, "_ddtrace_ids", lambda: None) + + assert obs_correlation() == {} + + def test_resolver_exception_is_swallowed(self, monkeypatch): + """A misbehaving tracer must not propagate out of obs_correlation.""" + monkeypatch.delenv("SGP_OBS_MODE", raising=False) # dd_only + + def boom(): + raise RuntimeError("tracer blew up") + + monkeypatch.setattr(obs_ids, "_ddtrace_ids", boom) + assert obs_correlation() == {} + + +class TestIdFormatting: + """Pin the W3C hex shape (32-hex trace, 16-hex span) of the resolvers.""" + + def test_ddtrace_ids_formats_w3c_hex(self, monkeypatch): + ctx = types.SimpleNamespace(trace_id=0xABC, span_id=0xFF) + tracer = types.SimpleNamespace(current_trace_context=lambda: ctx) + fake_ddtrace = types.ModuleType("ddtrace") + fake_trace = types.ModuleType("ddtrace.trace") + fake_trace.tracer = tracer + monkeypatch.setitem(sys.modules, "ddtrace", fake_ddtrace) + monkeypatch.setitem(sys.modules, "ddtrace.trace", fake_trace) + + trace_id, span_id = obs_ids._ddtrace_ids() + assert trace_id == "00000000000000000000000000000abc" + assert span_id == "000000000000000000ff"[-16:] # 16-hex + assert len(trace_id) == 32 and len(span_id) == 16 + + def test_lgtm_ids_formats_w3c_hex(self, monkeypatch): + span_ctx = types.SimpleNamespace(trace_id=0xABC, span_id=0xFF, is_valid=True) + current_span = types.SimpleNamespace(get_span_context=lambda: span_ctx) + fake_trace_mod = types.SimpleNamespace(get_current_span=lambda: current_span) + fake_otel = types.ModuleType("opentelemetry") + fake_otel.trace = fake_trace_mod + monkeypatch.setitem(sys.modules, "opentelemetry", fake_otel) + + trace_id, span_id = obs_ids._lgtm_ids() + assert trace_id == "00000000000000000000000000000abc" + assert len(trace_id) == 32 and len(span_id) == 16 + + def test_ddtrace_ids_none_when_no_context(self, monkeypatch): + tracer = types.SimpleNamespace(current_trace_context=lambda: None) + fake_ddtrace = types.ModuleType("ddtrace") + fake_trace = types.ModuleType("ddtrace.trace") + fake_trace.tracer = tracer + monkeypatch.setitem(sys.modules, "ddtrace", fake_ddtrace) + monkeypatch.setitem(sys.modules, "ddtrace.trace", fake_trace) + + assert obs_ids._ddtrace_ids() is None diff --git a/tests/lib/core/tracing/test_obs_span.py b/tests/lib/core/tracing/test_obs_span.py new file mode 100644 index 000000000..2638156a2 --- /dev/null +++ b/tests/lib/core/tracing/test_obs_span.py @@ -0,0 +1,399 @@ +from __future__ import annotations + +import sys +import types +from unittest.mock import MagicMock + +from agentex.lib.core.tracing import obs_span +from agentex.lib.core.tracing.trace import Trace + + +# --------------------------------------------------------------------------- # +# Fake OTel (lgtm) and fake ddtrace (dd_only) SDKs injected via sys.modules. +# --------------------------------------------------------------------------- # +class _FakeSpanContext: + def __init__(self, trace_id: int, span_id: int, is_valid: bool = True): + self.trace_id = trace_id + self.span_id = span_id + self.is_valid = is_valid + + +class _FakeOtelSpan: + def __init__(self, name: str, trace_id: int, span_id: int): + self.name = name + self._ctx = _FakeSpanContext(trace_id, span_id) + self.ended = False + self.attributes: dict = {} + + def set_attribute(self, key, value): + self.attributes[key] = value + + def get_span_context(self): + return self._ctx + + def end(self): + self.ended = True + + +def _install_fake_otel(monkeypatch, *, trace_id=0xABC, span_id=0xFF): + record: dict = {"span": None, "attached": [], "detached": []} + + def start_span(name): + span = _FakeOtelSpan(name, trace_id, span_id) + record["span"] = span + return span + + tracer = types.SimpleNamespace(start_span=start_span) + fake_trace = types.SimpleNamespace( + get_tracer=lambda _name: tracer, + set_span_in_context=lambda span: {"span": span}, + ) + fake_context = types.SimpleNamespace( + attach=lambda ctx: record["attached"].append(ctx) or object(), + detach=lambda token: record["detached"].append(token), + ) + fake_otel = types.ModuleType("opentelemetry") + fake_otel.trace = fake_trace + fake_otel.context = fake_context + monkeypatch.setitem(sys.modules, "opentelemetry", fake_otel) + return record + + +class _FakeDDSpan: + def __init__(self, name: str, trace_id: int, span_id: int): + self.name = name + self.trace_id = trace_id + self.span_id = span_id + self.finished = False + self.tags: dict = {} + + def set_tag(self, key, value): + self.tags[key] = value + + def finish(self): + self.finished = True + + +def _install_fake_ddtrace(monkeypatch, *, active=True, trace_id=0xABC, span_id=0xFF): + record: dict = {"span": None, "started": []} + + def start_span(name, activate=False): + span = _FakeDDSpan(name, trace_id, span_id) + record["span"] = span + record["started"].append((name, activate)) + return span + + tracer = types.SimpleNamespace( + current_trace_context=lambda: (object() if active else None), + start_span=start_span, + ) + fake_ddtrace = types.ModuleType("ddtrace") + fake_trace = types.ModuleType("ddtrace.trace") + fake_trace.tracer = tracer + monkeypatch.setitem(sys.modules, "ddtrace", fake_ddtrace) + monkeypatch.setitem(sys.modules, "ddtrace.trace", fake_trace) + return record + + +# --------------------------------------------------------------------------- # +# lgtm -> OTel wrapper +# --------------------------------------------------------------------------- # +class TestOtelWrapper: + def test_lgtm_opens_named_span_and_reads_its_ids(self, monkeypatch): + monkeypatch.setenv("SGP_OBS_MODE", "lgtm") + record = _install_fake_otel(monkeypatch, trace_id=0xABC, span_id=0xFF) + + handle = obs_span.open_obs_span( + "rocket.tool.fetch", business_span_id="bspan-1", business_trace_id="btrace-1" + ) + + assert handle is not None + assert record["span"].name == "rocket.tool.fetch" # named for the step + assert len(record["attached"]) == 1 # made active + assert handle.correlation == { + "obs_trace_id": "00000000000000000000000000000abc", + "obs_span_id": "000000000000000000ff"[-16:], + } + # reverse tag: business ids stamped on the obs span + assert record["span"].attributes == { + "agentex.business_span_id": "bspan-1", + "agentex.business_trace_id": "btrace-1", + } + + def test_invalid_span_context_yields_empty_correlation(self, monkeypatch): + monkeypatch.setenv("SGP_OBS_MODE", "lgtm") + _install_fake_otel(monkeypatch) + + def start_span(name): + span = _FakeOtelSpan(name, 0, 0) + span._ctx = _FakeSpanContext(0, 0, is_valid=False) + return span + + sys.modules["opentelemetry"].trace.get_tracer = lambda _n: types.SimpleNamespace( + start_span=start_span + ) + handle = obs_span.open_obs_span("step") + assert handle is not None + assert handle.correlation == {} + + def test_close_detaches_and_ends(self, monkeypatch): + monkeypatch.setenv("SGP_OBS_MODE", "lgtm") + record = _install_fake_otel(monkeypatch) + handle = obs_span.open_obs_span("step") + + obs_span.close_obs_span(handle) + + assert record["span"].ended is True + assert len(record["detached"]) == 1 + + def test_close_none_is_noop(self): + obs_span.close_obs_span(None) # must not raise + + +# --------------------------------------------------------------------------- # +# dd_only -> ddtrace wrapper (only when a request trace is active) +# --------------------------------------------------------------------------- # +class TestDdtraceWrapper: + def test_dd_only_with_active_ctx_opens_named_span(self, monkeypatch): + monkeypatch.setenv("SGP_OBS_MODE", "dd_only") + record = _install_fake_ddtrace(monkeypatch, active=True, trace_id=0xABC, span_id=0xFF) + + handle = obs_span.open_obs_span( + "rocket.tool.fetch", business_span_id="bspan-9", business_trace_id="btrace-9" + ) + + assert handle is not None + assert record["span"].name == "rocket.tool.fetch" + assert record["started"] == [("rocket.tool.fetch", True)] # activated + assert handle.correlation == { + "obs_trace_id": "00000000000000000000000000000abc", + "obs_span_id": "000000000000000000ff"[-16:], + } + # reverse tag on the ddtrace span + assert record["span"].tags == { + "agentex.business_span_id": "bspan-9", + "agentex.business_trace_id": "btrace-9", + } + + obs_span.close_obs_span(handle) + assert record["span"].finished is True + + def test_dd_only_without_active_ctx_returns_none(self, monkeypatch): + """Bare-uvicorn / no ddtrace-run: nothing active -> no orphan wrapper.""" + monkeypatch.setenv("SGP_OBS_MODE", "dd_only") + record = _install_fake_ddtrace(monkeypatch, active=False) + + assert obs_span.open_obs_span("step") is None + assert record["span"] is None # never created a span + + +# --------------------------------------------------------------------------- # +# End-to-end through Trace.start_span / end_span +# --------------------------------------------------------------------------- # +class TestTraceIntegration: + def test_lgtm_business_span_tagged_with_wrapper_ids(self, monkeypatch): + monkeypatch.setenv("SGP_OBS_MODE", "lgtm") + record = _install_fake_otel(monkeypatch, trace_id=0x111, span_id=0x222) + + trace = Trace(processors=[], client=MagicMock(), trace_id="task-run-1") + span = trace.start_span(name="chat_completion") + + assert record["span"].name == "chat_completion" # dedicated named span + assert span.data["obs_trace_id"] == "00000000000000000000000000000111" + assert span.data["obs_span_id"] == "0000000000000222" + assert span.trace_id == "task-run-1" # business id unchanged + assert span.id in trace._obs_handles + # bidirectional: the obs span carries the business ids (reverse tag), + # and the business span carries the obs ids (forward edge). + assert record["span"].attributes == { + "agentex.business_span_id": span.id, + "agentex.business_trace_id": "task-run-1", + } + + trace.end_span(span) + assert record["span"].ended is True + assert span.id not in trace._obs_handles + + def test_dd_only_business_span_tagged_via_ddtrace(self, monkeypatch): + monkeypatch.setenv("SGP_OBS_MODE", "dd_only") + record = _install_fake_ddtrace(monkeypatch, active=True, trace_id=0x111, span_id=0x222) + + trace = Trace(processors=[], client=MagicMock(), trace_id="task-run-2") + span = trace.start_span(name="get_state") + + assert record["span"].name == "get_state" + assert span.data["obs_trace_id"] == "00000000000000000000000000000111" + assert span.data["obs_span_id"] == "0000000000000222" + assert record["span"].tags == { + "agentex.business_span_id": span.id, + "agentex.business_trace_id": "task-run-2", + } + + trace.end_span(span) + assert record["span"].finished is True + assert span.id not in trace._obs_handles + + def test_dd_only_no_ctx_falls_back_to_ambient(self, monkeypatch): + monkeypatch.setenv("SGP_OBS_MODE", "dd_only") + _install_fake_ddtrace(monkeypatch, active=False) + monkeypatch.setattr("agentex.lib.core.tracing.trace.obs_correlation", lambda: {}) + + trace = Trace(processors=[], client=MagicMock(), trace_id="task-run-3") + span = trace.start_span(name="get_state") + + assert trace._obs_handles == {} # no wrapper opened + assert span.data is None # nothing tagged + trace.end_span(span) # must not raise + + +# --------------------------------------------------------------------------- # +# Non-interference: the two backends are mutually exclusive per mode. +# --------------------------------------------------------------------------- # +class TestNonInterference: + def test_lgtm_touches_only_otel(self, monkeypatch): + monkeypatch.setenv("SGP_OBS_MODE", "lgtm") + otel = _install_fake_otel(monkeypatch) + dd = _install_fake_ddtrace(monkeypatch, active=True) + + obs_span.open_obs_span("step") + + assert otel["span"] is not None # OTel wrapper opened + assert dd["span"] is None # ddtrace never touched + + def test_dd_only_touches_only_ddtrace(self, monkeypatch): + monkeypatch.setenv("SGP_OBS_MODE", "dd_only") + otel = _install_fake_otel(monkeypatch) + dd = _install_fake_ddtrace(monkeypatch, active=True) + + obs_span.open_obs_span("step") + + assert dd["span"] is not None # ddtrace wrapper opened + assert otel["span"] is None # OTel never touched + + +# --------------------------------------------------------------------------- # +# No-op when unconfigured, and never fails the app call. +# --------------------------------------------------------------------------- # +class TestNeverFails: + def test_lgtm_no_otel_installed_returns_none(self, monkeypatch): + monkeypatch.setenv("SGP_OBS_MODE", "lgtm") + monkeypatch.setitem(sys.modules, "opentelemetry", None) # import -> ImportError + assert obs_span.open_obs_span("step") is None + + def test_dd_only_no_ddtrace_installed_returns_none(self, monkeypatch): + monkeypatch.setenv("SGP_OBS_MODE", "dd_only") + monkeypatch.setitem(sys.modules, "ddtrace.trace", None) # import -> ImportError + assert obs_span.open_obs_span("step") is None + + def test_backend_exception_is_swallowed(self, monkeypatch): + monkeypatch.setenv("SGP_OBS_MODE", "lgtm") + _install_fake_otel(monkeypatch) + + def boom(_name): + raise RuntimeError("tracer blew up") + + sys.modules["opentelemetry"].trace.get_tracer = boom + assert obs_span.open_obs_span("step") is None # inner guard + + def test_top_level_guard_swallows_get_mode_error(self, monkeypatch): + # Even if mode resolution itself raises, open_obs_span must not. + monkeypatch.setattr(obs_span, "get_obs_mode", lambda: (_ for _ in ()).throw(RuntimeError())) + assert obs_span.open_obs_span("step") is None + + def test_close_swallows_closer_error(self): + handle = obs_span.ObsSpanHandle({}, lambda: (_ for _ in ()).throw(RuntimeError())) + obs_span.close_obs_span(handle) # must not raise + + def test_unconfigured_lgtm_yields_usable_span_no_raise(self, monkeypatch): + # lgtm requested but OTel not installed: the REAL open_obs_span returns + # None, obs_correlation() returns {} (also no tracer) -> the business + # span is created and fully usable, and nothing raised. + monkeypatch.setenv("SGP_OBS_MODE", "lgtm") + monkeypatch.setitem(sys.modules, "opentelemetry", None) + + trace = Trace(processors=[], client=MagicMock(), trace_id="task-run-4") + span = trace.start_span(name="safe") + + assert span.trace_id == "task-run-4" + assert trace._obs_handles == {} # no wrapper + trace.end_span(span) # must not raise + + +def _install_fake_otel_sequence(monkeypatch, *, trace_id: int, first_span_id: int): + """Fake OTel whose wrapper spans all share ``trace_id`` (children of the one + turn/request obs trace) but get sequential distinct span ids.""" + state: dict = {"next": first_span_id, "spans": []} + + def start_span(name): + sid = state["next"] + state["next"] += 1 + span = _FakeOtelSpan(name, trace_id, sid) + state["spans"].append(span) + return span + + tracer = types.SimpleNamespace(start_span=start_span) + fake_trace = types.SimpleNamespace( + get_tracer=lambda _name: tracer, + set_span_in_context=lambda span: {"span": span}, + ) + fake_context = types.SimpleNamespace( + attach=lambda ctx: object(), + detach=lambda token: None, + ) + fake_otel = types.ModuleType("opentelemetry") + fake_otel.trace = fake_trace + fake_otel.context = fake_context + monkeypatch.setitem(sys.modules, "opentelemetry", fake_otel) + return state + + +class TestTurn2Example: + """Maps the 3-turn mortgage example, Turn 2 (obs trace B): + + get_state -> wrapper wB1 -> obs_span_id = wB1 + retrieve_docs -> wrapper wB2 -> obs_span_id = wB2 + chat_completion -> wrapper wB3 -> obs_span_id = wB3 + create_message -> wrapper wB4 -> obs_span_id = wB4 + + Each step opens its OWN dedicated span named for the step; all four share the + one turn obs trace B, but obs_span_id is distinct per step (not all rB). + """ + + def test_turn2_each_step_gets_distinct_named_wrapper_under_trace_B(self, monkeypatch): + monkeypatch.setenv("SGP_OBS_MODE", "lgtm") + # Turn 2's request obs trace = B (0xB); wrappers get span ids 0xB1.. . + state = _install_fake_otel_sequence(monkeypatch, trace_id=0xB, first_span_id=0xB1) + + run_id = "task-run-mortgage" # business trace_id = the run/task id + trace = Trace(processors=[], client=MagicMock(), trace_id=run_id) + + steps = ["get_state", "retrieve_docs", "chat_completion", "create_message"] + business = [] + for step in steps: + with trace.span(name=step) as s: + business.append(s) + + obs_trace_B = format(0xB, "032x") + expected_obs_span = [format(sid, "016x") for sid in (0xB1, 0xB2, 0xB3, 0xB4)] + + # one dedicated wrapper per step, named for the step, in order + assert [w.name for w in state["spans"]] == steps + + for biz, wrapper, exp_span in zip(business, state["spans"], expected_obs_span): + # forward edge: business span carries the wrapper's ids + assert biz.data["obs_trace_id"] == obs_trace_B # all under trace B + assert biz.data["obs_span_id"] == exp_span # distinct wBn + # reverse tag: wrapper carries the business ids + assert wrapper.attributes == { + "agentex.business_span_id": biz.id, + "agentex.business_trace_id": run_id, + } + + # the whole point of the fix: obs_span_id is DISTINCT per step ... + obs_span_ids = [b.data["obs_span_id"] for b in business] + assert obs_span_ids == expected_obs_span + assert len(set(obs_span_ids)) == 4 + # ... while all four share the single turn obs trace B + assert {b.data["obs_trace_id"] for b in business} == {obs_trace_B} + # business trace stays the run/task id, not the obs trace + assert {b.trace_id for b in business} == {run_id} From 637d74f8c94e2403d6a053511aea458c9db2aeef Mon Sep 17 00:00:00 2001 From: Nitesh Dhanpal Date: Sat, 1 Aug 2026 12:29:32 -0700 Subject: [PATCH 2/4] observability: propagate business-span error status to wrapper obs span Observability review (P1): the dedicated wrapper obs span was ended/finished without recording failure, so a failed business step (e.g. chat_completion) showed green in Tempo/DD -- violating "observe both success and failure" and undercutting the meaningful-obs_span_id goal. close_obs_span now takes the business span's error (from get_span_error) and marks the obs span before closing: - OTel: span.set_status(Status(ERROR, msg)) + error.type attribute - ddtrace: span.error = 1 + error.type / error.message tags end_span passes error=get_span_error(span) on both sync and async paths. Success path is unchanged (no status set). Guarded so error-marking can never break the close. Co-Authored-By: Claude Opus 4.8 --- src/agentex/lib/core/tracing/obs_span.py | 48 ++++++++++++++--- src/agentex/lib/core/tracing/trace.py | 12 +++-- tests/lib/core/tracing/test_obs_span.py | 69 ++++++++++++++++++++++++ 3 files changed, 116 insertions(+), 13 deletions(-) diff --git a/src/agentex/lib/core/tracing/obs_span.py b/src/agentex/lib/core/tracing/obs_span.py index 77186d2ad..4d0b6c1db 100644 --- a/src/agentex/lib/core/tracing/obs_span.py +++ b/src/agentex/lib/core/tracing/obs_span.py @@ -47,7 +47,11 @@ class ObsSpanHandle: __slots__ = ("correlation", "_close") - def __init__(self, correlation: Dict[str, str], close: Callable[[], None]): + def __init__( + self, + correlation: Dict[str, str], + close: Callable[[Optional[Dict[str, str]]], None], + ): self.correlation = correlation self._close = close @@ -79,11 +83,21 @@ def _open_otel_span( sc = span.get_span_context() correlation = _hex_ids(sc.trace_id, sc.span_id) if (sc and sc.is_valid) else {} - def _close() -> None: + def _close(error: Optional[Dict[str, str]] = None) -> None: try: - context.detach(token) + if error: + # Reflect the business-step failure on the obs span so it + # isn't a false green when you pivot from a failed span. + span.set_status( + trace.Status(trace.StatusCode.ERROR, error.get("message")) + ) + if error.get("type"): + span.set_attribute("error.type", error["type"]) finally: - span.end() + try: + context.detach(token) + finally: + span.end() return ObsSpanHandle(correlation, _close) except Exception: # pragma: no cover - best-effort; never break the business span @@ -110,7 +124,20 @@ def _open_ddtrace_span( if business_trace_id: span.set_tag(_ATTR_BUSINESS_TRACE_ID, business_trace_id) correlation = _hex_ids(span.trace_id, span.span_id) if span.trace_id else {} - return ObsSpanHandle(correlation, span.finish) + + def _close(error: Optional[Dict[str, str]] = None) -> None: + try: + if error: + # Reflect the business-step failure on the obs span. + span.error = 1 + if error.get("type"): + span.set_tag("error.type", error["type"]) + if error.get("message"): + span.set_tag("error.message", error["message"]) + finally: + span.finish() + + return ObsSpanHandle(correlation, _close) except Exception: # pragma: no cover - best-effort return None @@ -143,11 +170,16 @@ def open_obs_span( return None -def close_obs_span(handle: Optional[ObsSpanHandle]) -> None: - """Close the wrapper span (detach + end, or finish). Safe on ``None``.""" +def close_obs_span( + handle: Optional[ObsSpanHandle], + error: Optional[Dict[str, str]] = None, +) -> None: + """Close the wrapper span (detach + end, or finish). When ``error`` is given + (the business span failed), mark the obs span errored first so it reflects + failure rather than a false green. Safe on ``None``.""" if handle is None: return try: - handle._close() + handle._close(error) except Exception: # pragma: no cover - best-effort pass diff --git a/src/agentex/lib/core/tracing/trace.py b/src/agentex/lib/core/tracing/trace.py index 031e69e9c..7ac6aa2ef 100644 --- a/src/agentex/lib/core/tracing/trace.py +++ b/src/agentex/lib/core/tracing/trace.py @@ -17,7 +17,7 @@ close_obs_span, open_obs_span, ) -from agentex.lib.core.tracing.span_error import set_span_error +from agentex.lib.core.tracing.span_error import get_span_error, set_span_error from agentex.lib.core.tracing.span_queue import ( SpanEventType, AsyncSpanQueue, @@ -135,8 +135,9 @@ def end_span( if span.end_time is None: span.end_time = datetime.now(UTC) - # Close the dedicated obs wrapper span (detach context + end it). - close_obs_span(self._obs_handles.pop(span.id, None)) + # Close the dedicated obs wrapper span; propagate the business-span error + # (if any) so the obs span reflects failure, not a false green. + close_obs_span(self._obs_handles.pop(span.id, None), error=get_span_error(span)) span.input = recursive_model_dump(span.input) if span.input else None span.output = recursive_model_dump(span.output) if span.output else None @@ -304,8 +305,9 @@ async def end_span( if span.end_time is None: span.end_time = datetime.now(UTC) - # Close the dedicated obs wrapper span (detach context + end it). - close_obs_span(self._obs_handles.pop(span.id, None)) + # Close the dedicated obs wrapper span; propagate the business-span error + # (if any) so the obs span reflects failure, not a false green. + close_obs_span(self._obs_handles.pop(span.id, None), error=get_span_error(span)) span.input = recursive_model_dump(span.input) if span.input else None span.output = recursive_model_dump(span.output) if span.output else None diff --git a/tests/lib/core/tracing/test_obs_span.py b/tests/lib/core/tracing/test_obs_span.py index 2638156a2..aed5e2abc 100644 --- a/tests/lib/core/tracing/test_obs_span.py +++ b/tests/lib/core/tracing/test_obs_span.py @@ -4,6 +4,8 @@ import types from unittest.mock import MagicMock +import pytest + from agentex.lib.core.tracing import obs_span from agentex.lib.core.tracing.trace import Trace @@ -18,16 +20,30 @@ def __init__(self, trace_id: int, span_id: int, is_valid: bool = True): self.is_valid = is_valid +class _FakeStatusCode: + ERROR = "ERROR" + OK = "OK" + UNSET = "UNSET" + + +def _FakeStatus(code, description=None): + return {"code": code, "description": description} + + class _FakeOtelSpan: def __init__(self, name: str, trace_id: int, span_id: int): self.name = name self._ctx = _FakeSpanContext(trace_id, span_id) self.ended = False self.attributes: dict = {} + self.status = None def set_attribute(self, key, value): self.attributes[key] = value + def set_status(self, status): + self.status = status + def get_span_context(self): return self._ctx @@ -47,6 +63,8 @@ def start_span(name): fake_trace = types.SimpleNamespace( get_tracer=lambda _name: tracer, set_span_in_context=lambda span: {"span": span}, + Status=_FakeStatus, + StatusCode=_FakeStatusCode, ) fake_context = types.SimpleNamespace( attach=lambda ctx: record["attached"].append(ctx) or object(), @@ -65,6 +83,7 @@ def __init__(self, name: str, trace_id: int, span_id: int): self.trace_id = trace_id self.span_id = span_id self.finished = False + self.error = 0 self.tags: dict = {} def set_tag(self, key, value): @@ -149,6 +168,27 @@ def test_close_detaches_and_ends(self, monkeypatch): def test_close_none_is_noop(self): obs_span.close_obs_span(None) # must not raise + def test_close_with_error_marks_otel_status(self, monkeypatch): + monkeypatch.setenv("SGP_OBS_MODE", "lgtm") + record = _install_fake_otel(monkeypatch) + handle = obs_span.open_obs_span("step") + + obs_span.close_obs_span(handle, error={"type": "ValueError", "message": "boom"}) + + assert record["span"].status == {"code": "ERROR", "description": "boom"} + assert record["span"].attributes.get("error.type") == "ValueError" + assert record["span"].ended is True + + def test_close_without_error_leaves_otel_status_unset(self, monkeypatch): + monkeypatch.setenv("SGP_OBS_MODE", "lgtm") + record = _install_fake_otel(monkeypatch) + handle = obs_span.open_obs_span("step") + + obs_span.close_obs_span(handle) # success path + + assert record["span"].status is None + assert record["span"].ended is True + # --------------------------------------------------------------------------- # # dd_only -> ddtrace wrapper (only when a request trace is active) @@ -186,6 +226,18 @@ def test_dd_only_without_active_ctx_returns_none(self, monkeypatch): assert obs_span.open_obs_span("step") is None assert record["span"] is None # never created a span + def test_close_with_error_marks_ddtrace_span(self, monkeypatch): + monkeypatch.setenv("SGP_OBS_MODE", "dd_only") + record = _install_fake_ddtrace(monkeypatch, active=True) + handle = obs_span.open_obs_span("step") + + obs_span.close_obs_span(handle, error={"type": "ValueError", "message": "boom"}) + + assert record["span"].error == 1 + assert record["span"].tags.get("error.type") == "ValueError" + assert record["span"].tags.get("error.message") == "boom" + assert record["span"].finished is True + # --------------------------------------------------------------------------- # # End-to-end through Trace.start_span / end_span @@ -233,6 +285,21 @@ def test_dd_only_business_span_tagged_via_ddtrace(self, monkeypatch): assert record["span"].finished is True assert span.id not in trace._obs_handles + def test_lgtm_wrapper_marked_error_when_business_step_raises(self, monkeypatch): + monkeypatch.setenv("SGP_OBS_MODE", "lgtm") + record = _install_fake_otel(monkeypatch, trace_id=0x111, span_id=0x222) + + trace = Trace(processors=[], client=MagicMock(), trace_id="task-run-err") + with pytest.raises(ValueError): + with trace.span(name="chat_completion"): + raise ValueError("boom") + + # the failed step's obs span reflects the failure, not a false green + assert record["span"].name == "chat_completion" + assert record["span"].ended is True + assert record["span"].status == {"code": "ERROR", "description": "boom"} + assert record["span"].attributes.get("error.type") == "ValueError" + def test_dd_only_no_ctx_falls_back_to_ambient(self, monkeypatch): monkeypatch.setenv("SGP_OBS_MODE", "dd_only") _install_fake_ddtrace(monkeypatch, active=False) @@ -335,6 +402,8 @@ def start_span(name): fake_trace = types.SimpleNamespace( get_tracer=lambda _name: tracer, set_span_in_context=lambda span: {"span": span}, + Status=_FakeStatus, + StatusCode=_FakeStatusCode, ) fake_context = types.SimpleNamespace( attach=lambda ctx: object(), From 5658801c8355eed0a77a9423c02998a0a157a7ea Mon Sep 17 00:00:00 2001 From: Nitesh Dhanpal Date: Sat, 1 Aug 2026 22:36:59 -0700 Subject: [PATCH 3/4] fix(tracing): nest ddtrace wrapper under the active context (child_of) ddtrace start_span does not auto-parent (unlike OTel): start_span(name) mints a new ROOT trace every call, so a turn's business spans scattered across N Datadog traces (verified live: 52 spans -> 52 distinct obs_trace_ids). Pass child_of=current_trace_context() so wrappers nest under the request/turn trace and roll up into one trace; obs_span_id stays distinct per step. Co-Authored-By: Claude Opus 4.8 --- src/agentex/lib/core/tracing/obs_span.py | 10 ++++++++-- tests/lib/core/tracing/test_obs_span.py | 15 +++++++++++---- 2 files changed, 19 insertions(+), 6 deletions(-) diff --git a/src/agentex/lib/core/tracing/obs_span.py b/src/agentex/lib/core/tracing/obs_span.py index 4d0b6c1db..39234b8ff 100644 --- a/src/agentex/lib/core/tracing/obs_span.py +++ b/src/agentex/lib/core/tracing/obs_span.py @@ -116,9 +116,15 @@ def _open_ddtrace_span( try: # Only wrap when ddtrace is actually tracing the request; otherwise a # wrapper would be an orphan root trace in an un-instrumented process. - if tracer.current_trace_context() is None: + ctx = tracer.current_trace_context() + if ctx is None: return None - span = tracer.start_span(name, activate=True) + # child_of=ctx is load-bearing: ddtrace's start_span does NOT auto-parent + # to the active span (unlike OTel), so start_span(name) alone mints a NEW + # root trace every call -- scattering a turn's business spans across N + # Datadog traces. Parenting to the active request/turn context rolls them + # into one trace while obs_span_id stays distinct per step. + span = tracer.start_span(name, child_of=ctx, activate=True) if business_span_id: span.set_tag(_ATTR_BUSINESS_SPAN_ID, business_span_id) if business_trace_id: diff --git a/tests/lib/core/tracing/test_obs_span.py b/tests/lib/core/tracing/test_obs_span.py index aed5e2abc..c35fbe1a0 100644 --- a/tests/lib/core/tracing/test_obs_span.py +++ b/tests/lib/core/tracing/test_obs_span.py @@ -95,15 +95,17 @@ def finish(self): def _install_fake_ddtrace(monkeypatch, *, active=True, trace_id=0xABC, span_id=0xFF): record: dict = {"span": None, "started": []} + ctx_obj = object() if active else None + record["ctx"] = ctx_obj - def start_span(name, activate=False): + def start_span(name, child_of=None, activate=False): span = _FakeDDSpan(name, trace_id, span_id) record["span"] = span - record["started"].append((name, activate)) + record["started"].append({"name": name, "child_of": child_of, "activate": activate}) return span tracer = types.SimpleNamespace( - current_trace_context=lambda: (object() if active else None), + current_trace_context=lambda: ctx_obj, start_span=start_span, ) fake_ddtrace = types.ModuleType("ddtrace") @@ -204,7 +206,12 @@ def test_dd_only_with_active_ctx_opens_named_span(self, monkeypatch): assert handle is not None assert record["span"].name == "rocket.tool.fetch" - assert record["started"] == [("rocket.tool.fetch", True)] # activated + started = record["started"][0] + assert started["name"] == "rocket.tool.fetch" + assert started["activate"] is True + # child_of is the active request/turn context -> the wrapper nests under + # it instead of minting a new root trace (ddtrace does not auto-parent). + assert started["child_of"] is record["ctx"] assert handle.correlation == { "obs_trace_id": "00000000000000000000000000000abc", "obs_span_id": "000000000000000000ff"[-16:], From 51a4d37e1e0012e77bbe394e5f9a8472cc303280 Mon Sep 17 00:00:00 2001 From: Nitesh Dhanpal Date: Sun, 2 Aug 2026 16:01:48 -0700 Subject: [PATCH 4/4] fix(tracing): track obs wrapper spans in a module-level registry The obs wrapper span (opened in start_span, ended in end_span) was tracked in an INSTANCE dict (self._obs_handles). But TracingService creates a fresh Trace object for every call -- self._tracer.trace(trace_id) in BOTH start_span and end_span -- so end_span ran on a different instance with an empty dict: the handle was never found, close_obs_span(None) was a no-op, and the OTel/ddtrace wrapper span was never .end()ed. Consequence in lgtm mode: the wrapper span records and its ids are written to Postgres (read at start), but since Simple/Batch span processors only export on span end, the span never reaches Tempo -- the turn trace was silently missing while everything looked correct (provider ours, sampler ALWAYS_ON, recording=True, ids stored). Fix: move the handle registry to module level, keyed by the uuid4 span id, so it survives across Trace instances. Adds a regression test that starts a span on one Trace instance and ends it on another. Co-Authored-By: Claude Opus 4.8 --- src/agentex/lib/core/tracing/trace.py | 29 +++++++++++----- tests/lib/core/tracing/test_obs_span.py | 45 ++++++++++++++++++++++--- 2 files changed, 61 insertions(+), 13 deletions(-) diff --git a/src/agentex/lib/core/tracing/trace.py b/src/agentex/lib/core/tracing/trace.py index 7ac6aa2ef..b8c3a6fc0 100644 --- a/src/agentex/lib/core/tracing/trace.py +++ b/src/agentex/lib/core/tracing/trace.py @@ -30,6 +30,17 @@ logger = make_logger(__name__) +# Live per-business-span obs wrapper spans, keyed by the (uuid4) business span id, +# in a MODULE-LEVEL registry -- deliberately NOT on the Trace/AsyncTrace instance. +# TracingService creates a FRESH trace object for every call +# (`self._tracer.trace(trace_id)` in both start_span and end_span), so an +# instance-local dict loses the handle between start and end: end_span's new +# instance can't find it, close_obs_span(None) is a no-op, and the OTel wrapper +# span is never .end()ed -> never exported (Simple/Batch processors only emit on +# end). A module-level dict keyed by the unique span id survives across instances; +# uuid4 span ids cannot collide across concurrent traces. +_OBS_HANDLES: dict[str, ObsSpanHandle] = {} + class Trace: """ @@ -54,8 +65,9 @@ def __init__( self.processors = processors self.client = client self.trace_id = trace_id - # Live per-business-span obs wrapper spans, keyed by business span id. - self._obs_handles: dict[str, ObsSpanHandle] = {} + # Obs wrapper spans are tracked in the module-level _OBS_HANDLES registry + # (see comment there): a fresh trace object is created per start/end call, + # so the handle must not live on the instance. def start_span( self, @@ -112,7 +124,7 @@ def start_span( task_id=task_id, ) if obs_handle is not None: - self._obs_handles[span.id] = obs_handle + _OBS_HANDLES[span.id] = obs_handle for processor in self.processors: processor.on_span_start(span) @@ -137,7 +149,7 @@ def end_span( # Close the dedicated obs wrapper span; propagate the business-span error # (if any) so the obs span reflects failure, not a false green. - close_obs_span(self._obs_handles.pop(span.id, None), error=get_span_error(span)) + close_obs_span(_OBS_HANDLES.pop(span.id, None), error=get_span_error(span)) span.input = recursive_model_dump(span.input) if span.input else None span.output = recursive_model_dump(span.output) if span.output else None @@ -225,8 +237,9 @@ def __init__( self.client = client self.trace_id = trace_id self._span_queue = span_queue or get_default_span_queue() - # Live per-business-span obs wrapper spans, keyed by business span id. - self._obs_handles: dict[str, ObsSpanHandle] = {} + # Obs wrapper spans are tracked in the module-level _OBS_HANDLES registry + # (see comment there): a fresh trace object is created per start/end call, + # so the handle must not live on the instance. async def start_span( self, @@ -282,7 +295,7 @@ async def start_span( task_id=task_id, ) if obs_handle is not None: - self._obs_handles[span.id] = obs_handle + _OBS_HANDLES[span.id] = obs_handle if self.processors: self._span_queue.enqueue(SpanEventType.START, span.model_copy(deep=True), self.processors) @@ -307,7 +320,7 @@ async def end_span( # Close the dedicated obs wrapper span; propagate the business-span error # (if any) so the obs span reflects failure, not a false green. - close_obs_span(self._obs_handles.pop(span.id, None), error=get_span_error(span)) + close_obs_span(_OBS_HANDLES.pop(span.id, None), error=get_span_error(span)) span.input = recursive_model_dump(span.input) if span.input else None span.output = recursive_model_dump(span.output) if span.output else None diff --git a/tests/lib/core/tracing/test_obs_span.py b/tests/lib/core/tracing/test_obs_span.py index c35fbe1a0..aa7d826bb 100644 --- a/tests/lib/core/tracing/test_obs_span.py +++ b/tests/lib/core/tracing/test_obs_span.py @@ -7,9 +7,20 @@ import pytest from agentex.lib.core.tracing import obs_span +from agentex.lib.core.tracing import trace as trace_module from agentex.lib.core.tracing.trace import Trace +@pytest.fixture(autouse=True) +def _clear_obs_handles(): + """The obs-handle registry is module-level (survives across Trace instances, + which is the whole point of the fix). Clear it around each test so leftover + handles never leak between tests.""" + trace_module._OBS_HANDLES.clear() + yield + trace_module._OBS_HANDLES.clear() + + # --------------------------------------------------------------------------- # # Fake OTel (lgtm) and fake ddtrace (dd_only) SDKs injected via sys.modules. # --------------------------------------------------------------------------- # @@ -261,7 +272,7 @@ def test_lgtm_business_span_tagged_with_wrapper_ids(self, monkeypatch): assert span.data["obs_trace_id"] == "00000000000000000000000000000111" assert span.data["obs_span_id"] == "0000000000000222" assert span.trace_id == "task-run-1" # business id unchanged - assert span.id in trace._obs_handles + assert span.id in trace_module._OBS_HANDLES # bidirectional: the obs span carries the business ids (reverse tag), # and the business span carries the obs ids (forward edge). assert record["span"].attributes == { @@ -271,7 +282,31 @@ def test_lgtm_business_span_tagged_with_wrapper_ids(self, monkeypatch): trace.end_span(span) assert record["span"].ended is True - assert span.id not in trace._obs_handles + assert span.id not in trace_module._OBS_HANDLES + + def test_wrapper_ends_across_separate_trace_instances(self, monkeypatch): + # Regression for the export bug: TracingService creates a FRESH trace + # object for start_span AND for end_span (self._tracer.trace(trace_id) in + # both). The obs handle is stored in the module-level registry, so a + # DIFFERENT instance ending the span still finds it and calls .end() on + # the OTel wrapper. With an instance-local dict this regressed: end_span's + # new instance had an empty dict -> close_obs_span(None) -> the wrapper + # span was never ended -> never exported to Tempo (recording, ids stored, + # but absent from the trace backend). + monkeypatch.setenv("SGP_OBS_MODE", "lgtm") + record = _install_fake_otel(monkeypatch, trace_id=0x111, span_id=0x222) + + starter = Trace(processors=[], client=MagicMock(), trace_id="task-run-x") + span = starter.start_span(name="chat_completion") + assert record["span"].ended is False + assert span.id in trace_module._OBS_HANDLES + + # A completely separate Trace instance ends the span. + ender = Trace(processors=[], client=MagicMock(), trace_id="task-run-x") + ender.end_span(span) + + assert record["span"].ended is True # wrapper WAS ended -> exportable + assert span.id not in trace_module._OBS_HANDLES # handle cleaned up def test_dd_only_business_span_tagged_via_ddtrace(self, monkeypatch): monkeypatch.setenv("SGP_OBS_MODE", "dd_only") @@ -290,7 +325,7 @@ def test_dd_only_business_span_tagged_via_ddtrace(self, monkeypatch): trace.end_span(span) assert record["span"].finished is True - assert span.id not in trace._obs_handles + assert span.id not in trace_module._OBS_HANDLES def test_lgtm_wrapper_marked_error_when_business_step_raises(self, monkeypatch): monkeypatch.setenv("SGP_OBS_MODE", "lgtm") @@ -315,7 +350,7 @@ def test_dd_only_no_ctx_falls_back_to_ambient(self, monkeypatch): trace = Trace(processors=[], client=MagicMock(), trace_id="task-run-3") span = trace.start_span(name="get_state") - assert trace._obs_handles == {} # no wrapper opened + assert span.id not in trace_module._OBS_HANDLES # no wrapper opened assert span.data is None # nothing tagged trace.end_span(span) # must not raise @@ -389,7 +424,7 @@ def test_unconfigured_lgtm_yields_usable_span_no_raise(self, monkeypatch): span = trace.start_span(name="safe") assert span.trace_id == "task-run-4" - assert trace._obs_handles == {} # no wrapper + assert span.id not in trace_module._OBS_HANDLES # no wrapper trace.end_span(span) # must not raise