From 7b34987f4ad47a542e1dc538001e9dfd7a1b262b Mon Sep 17 00:00:00 2001 From: Nitesh Dhanpal Date: Sun, 2 Aug 2026 22:18:57 -0700 Subject: [PATCH] feat(tracing): propagate OTel trace context across Temporal boundaries Temporal serializes start_workflow / execute_activity across (potentially cross-process) boundaries and does not carry the active W3C traceparent, so spans created inside a workflow or activity become detached roots -- the trace shatters at every Temporal hop. This bites agentex directly: adk.tracing.span creates the business span as a Temporal activity when in_temporal_workflow(), so without propagation those spans detach from the turn's obs trace. Wire temporalio's first-party TracingInterceptor onto both Temporal client factories (worker client + the ACP's workflow-starting TemporalClient) and the AgentexWorker, so client -> workflow -> activity is one trace. The interceptor injects context on outbound calls and extracts + roots execution spans under it, using the global OpenTelemetry propagator. - ENABLED BY DEFAULT. Opt out with AGENTEX_TEMPORAL_TRACE_INTERCEPTOR_ENABLED=false (also 0/no/off). Safe no-op (never raises) if temporalio's OTel contrib isn't importable, so default-on can't break a worker. - Tracing interceptor is placed OUTERMOST on the worker so existing business interceptors (and their spans) nest under the propagated span. Tests: tests/lib/core/tracing/test_temporal_interceptor.py -- default-on returns a TracingInterceptor, env opt-out returns [], contrib-missing returns []. Co-Authored-By: Claude Opus 4.8 --- .../lib/core/clients/temporal/utils.py | 5 ++ .../lib/core/temporal/workers/worker.py | 8 +- src/agentex/lib/core/tracing/temporal.py | 73 +++++++++++++++++++ .../core/tracing/test_temporal_interceptor.py | 40 ++++++++++ 4 files changed, 125 insertions(+), 1 deletion(-) create mode 100644 src/agentex/lib/core/tracing/temporal.py create mode 100644 tests/lib/core/tracing/test_temporal_interceptor.py diff --git a/src/agentex/lib/core/clients/temporal/utils.py b/src/agentex/lib/core/clients/temporal/utils.py index 95319720a..15b08cec6 100644 --- a/src/agentex/lib/core/clients/temporal/utils.py +++ b/src/agentex/lib/core/clients/temporal/utils.py @@ -9,6 +9,8 @@ from temporalio.converter import PayloadCodec, DataConverter from temporalio.contrib.pydantic import pydantic_data_converter +from agentex.lib.core.tracing.temporal import temporal_tracing_interceptors + # class DateTimeJSONEncoder(AdvancedJSONEncoder): # def default(self, o: Any) -> Any: # if isinstance(o, datetime.datetime): @@ -136,6 +138,9 @@ async def get_temporal_client( connect_kwargs: dict[str, Any] = { "target_host": temporal_address, "plugins": plugins, + # Propagate OTel trace context on outbound start_workflow / execute_activity + # (enabled by default; AGENTEX_TEMPORAL_TRACE_INTERCEPTOR_ENABLED=false to disable). + "interceptors": temporal_tracing_interceptors(), } if data_converter is not None: diff --git a/src/agentex/lib/core/temporal/workers/worker.py b/src/agentex/lib/core/temporal/workers/worker.py index 2b4958b1f..0cfe01185 100644 --- a/src/agentex/lib/core/temporal/workers/worker.py +++ b/src/agentex/lib/core/temporal/workers/worker.py @@ -29,6 +29,7 @@ from agentex.lib.utils.logging import make_logger from agentex.lib.utils.registration import register_agent +from agentex.lib.core.tracing.temporal import temporal_tracing_interceptors from agentex.lib.environment_variables import EnvironmentVariables from agentex.lib.core.compat.version_guard import assert_backend_compatible @@ -126,6 +127,9 @@ async def get_temporal_client( connect_kwargs: dict[str, Any] = { "target_host": temporal_address, "plugins": plugins, + # Propagate OTel trace context on outbound start_workflow / execute_activity + # (enabled by default; AGENTEX_TEMPORAL_TRACE_INTERCEPTOR_ENABLED=false to disable). + "interceptors": temporal_tracing_interceptors(), } if data_converter is not None: @@ -229,7 +233,9 @@ async def run( max_concurrent_activities=self.max_concurrent_activities, build_id=str(uuid.uuid4()), debug_mode=debug_enabled, # Disable deadlock detection in debug mode - interceptors=self.interceptors, # Pass interceptors to Worker + # Tracing interceptor OUTERMOST so business interceptors (and the spans + # they create) nest under the propagated workflow/activity span. + interceptors=[*temporal_tracing_interceptors(), *self.interceptors], ) logger.info(f"Starting workers for task queue: {self.task_queue}") diff --git a/src/agentex/lib/core/tracing/temporal.py b/src/agentex/lib/core/tracing/temporal.py new file mode 100644 index 000000000..484abc26b --- /dev/null +++ b/src/agentex/lib/core/tracing/temporal.py @@ -0,0 +1,73 @@ +"""OpenTelemetry trace-context propagation across Temporal boundaries. + +Temporal serializes ``start_workflow`` / ``execute_activity`` across (potentially +cross-process) boundaries, and does NOT carry the active W3C ``traceparent`` by +default. So any span created inside a workflow or activity becomes a **new +detached root** -- the trace shatters at every Temporal hop. + +This bites agentex directly: ``adk.tracing.span`` runs span creation as a +Temporal activity when ``in_temporal_workflow()`` is true, so without propagation +those business spans detach from the turn's obs trace. + +Wiring temporalio's first-party ``TracingInterceptor`` onto the Temporal client +and worker injects the active span context into Temporal headers on the caller +side and extracts + continues it on the workflow/activity side, using the global +OpenTelemetry propagator -- so ``client -> workflow -> activity`` is one trace. + +Enabled by DEFAULT. Set ``AGENTEX_TEMPORAL_TRACE_INTERCEPTOR_ENABLED=false`` +(also accepts ``0`` / ``no`` / ``off``) to turn it off. It also degrades to a +no-op -- and never raises -- if temporalio's OpenTelemetry contrib isn't +importable, so enabling it by default can't break a worker. +""" + +from __future__ import annotations + +import os +from typing import Any + +from agentex.lib.utils.logging import make_logger + +logger = make_logger(__name__) + +_ENABLE_ENV = "AGENTEX_TEMPORAL_TRACE_INTERCEPTOR_ENABLED" +_FALSEY = {"0", "false", "no", "off"} + + +def temporal_trace_interceptor_enabled() -> bool: + """Whether the Temporal OTel trace interceptor should be installed. + + Defaults to True; disabled only when ``AGENTEX_TEMPORAL_TRACE_INTERCEPTOR_ENABLED`` + is set to a falsy value (``0`` / ``false`` / ``no`` / ``off``).""" + return os.environ.get(_ENABLE_ENV, "true").strip().lower() not in _FALSEY + + +def temporal_tracing_interceptors() -> list[Any]: + """Interceptors that propagate OpenTelemetry trace context across Temporal. + + Returns ``[TracingInterceptor()]`` (enabled by default) so callers can splat + it into a client's / worker's ``interceptors=`` list. Returns ``[]`` when + disabled via env, or when temporalio's OpenTelemetry contrib is not + importable. Never raises -- observability wiring must not break a worker. + + ``TracingInterceptor`` implements both the client and worker interceptor + interfaces, so the same call is used on both sides: + - on the **client**, it injects context on outbound ``start_workflow`` / + ``execute_activity`` calls; + - on the **worker**, it extracts context and roots the workflow / activity + execution spans under it. + """ + if not temporal_trace_interceptor_enabled(): + logger.info("Temporal OTel trace interceptor disabled via %s", _ENABLE_ENV) + return [] + try: + from temporalio.contrib.opentelemetry import TracingInterceptor + + # Construct inside the try so a constructor failure (not just a missing + # contrib) also falls back to a no-op instead of aborting worker startup. + return [TracingInterceptor()] + except Exception as exc: # contrib unavailable OR constructor failure -> no-op, never raise + logger.warning( + "Temporal OTel trace interceptor unavailable (%s); traces will not propagate across Temporal boundaries.", + exc, + ) + return [] diff --git a/tests/lib/core/tracing/test_temporal_interceptor.py b/tests/lib/core/tracing/test_temporal_interceptor.py new file mode 100644 index 000000000..83c0c3681 --- /dev/null +++ b/tests/lib/core/tracing/test_temporal_interceptor.py @@ -0,0 +1,40 @@ +"""Unit tests for the Temporal OTel trace-interceptor wiring. + +Verifies the interceptor is on by default, the opt-out env flag, and the safe +no-op fallback when temporalio's OpenTelemetry contrib isn't importable. +""" + +import sys + +import pytest + +from agentex.lib.core.tracing import temporal as temporal_tracing + + +class TestTemporalTraceInterceptor: + def test_enabled_by_default(self, monkeypatch): + monkeypatch.delenv("AGENTEX_TEMPORAL_TRACE_INTERCEPTOR_ENABLED", raising=False) + assert temporal_tracing.temporal_trace_interceptor_enabled() is True + + interceptors = temporal_tracing.temporal_tracing_interceptors() + assert len(interceptors) == 1 + # temporalio's first-party OTel interceptor + assert type(interceptors[0]).__name__ == "TracingInterceptor" + + @pytest.mark.parametrize("value", ["false", "0", "no", "off", "FALSE", "Off"]) + def test_disabled_via_env(self, monkeypatch, value): + monkeypatch.setenv("AGENTEX_TEMPORAL_TRACE_INTERCEPTOR_ENABLED", value) + assert temporal_tracing.temporal_trace_interceptor_enabled() is False + assert temporal_tracing.temporal_tracing_interceptors() == [] + + @pytest.mark.parametrize("value", ["true", "1", "yes", "TRUE", "anything"]) + def test_enabled_for_non_falsy_values(self, monkeypatch, value): + monkeypatch.setenv("AGENTEX_TEMPORAL_TRACE_INTERCEPTOR_ENABLED", value) + assert temporal_tracing.temporal_trace_interceptor_enabled() is True + + def test_no_op_when_contrib_unimportable(self, monkeypatch): + # Enabled, but temporalio's OTel contrib not importable -> [] (never raises), + # so default-on can't break a worker that lacks the contrib. + monkeypatch.delenv("AGENTEX_TEMPORAL_TRACE_INTERCEPTOR_ENABLED", raising=False) + monkeypatch.setitem(sys.modules, "temporalio.contrib.opentelemetry", None) + assert temporal_tracing.temporal_tracing_interceptors() == []