From cbcd20aa4af76baa4410c949212e2515f16d20ec Mon Sep 17 00:00:00 2001 From: Nitesh Dhanpal Date: Tue, 21 Jul 2026 13:40:50 -0700 Subject: [PATCH] feat(tracing): propagate trace context into Temporal via OTel interceptor Attach temporalio.contrib.opentelemetry.TracingInterceptor to the Temporal client (both get_temporal_client entrypoints) when an OTel observability mode is active (SGP_OBS_MODE in {dual, lgtm}). This threads the caller's W3C trace context across the workflow -> activity boundary, so business spans created inside the model-loop activity see an active observability context and can be tagged with its trace_id via obs_correlation(). Gated + lazy-imported: default dd_only mode adds nothing and the temporalio OTel contrib is only imported when needed, so this is a no-op unless opted in. Co-Authored-By: Claude Opus 4.8 --- .../lib/core/clients/temporal/utils.py | 24 +++++++++++++++++++ .../lib/core/temporal/workers/worker.py | 22 +++++++++++++++++ 2 files changed, 46 insertions(+) diff --git a/src/agentex/lib/core/clients/temporal/utils.py b/src/agentex/lib/core/clients/temporal/utils.py index 95319720a..886b1a1fa 100644 --- a/src/agentex/lib/core/clients/temporal/utils.py +++ b/src/agentex/lib/core/clients/temporal/utils.py @@ -81,6 +81,26 @@ def validate_worker_interceptors(interceptors: list[Any]) -> None: ) +def _build_otel_interceptors() -> list: + """Return an OpenTelemetry Temporal interceptor when an OTel observability + mode is active (SGP_OBS_MODE in {dual, lgtm}), so trace context propagates + across the workflow -> activity boundary. This lets business spans created + inside the model-loop activity adopt the caller's observability trace_id. + + Empty in the default dd_only mode, or when the temporalio OTel contrib is + unavailable (safe no-op). + """ + import os + + if os.getenv("SGP_OBS_MODE", "").strip().lower() not in ("dual", "lgtm"): + return [] + try: + from temporalio.contrib.opentelemetry import TracingInterceptor + except ImportError: + return [] + return [TracingInterceptor()] + + async def get_temporal_client( temporal_address: str, metrics_url: str | None = None, @@ -146,6 +166,10 @@ async def get_temporal_client( dc = dataclasses.replace(dc, payload_codec=payload_codec) connect_kwargs["data_converter"] = dc + otel_interceptors = _build_otel_interceptors() + if otel_interceptors: + connect_kwargs["interceptors"] = otel_interceptors + if not metrics_url: client = await Client.connect(**connect_kwargs) else: diff --git a/src/agentex/lib/core/temporal/workers/worker.py b/src/agentex/lib/core/temporal/workers/worker.py index 2b4958b1f..fb4effac2 100644 --- a/src/agentex/lib/core/temporal/workers/worker.py +++ b/src/agentex/lib/core/temporal/workers/worker.py @@ -91,6 +91,24 @@ def _validate_interceptors(interceptors: list) -> None: ) +def _build_otel_interceptors() -> list: + """Return an OpenTelemetry Temporal interceptor when an OTel observability + mode is active (SGP_OBS_MODE in {dual, lgtm}), so trace context propagates + across the workflow -> activity boundary. This lets business spans created + inside the model-loop activity adopt the caller's observability trace_id. + + Empty in the default dd_only mode, or when the temporalio OTel contrib is + unavailable (safe no-op). + """ + if os.getenv("SGP_OBS_MODE", "").strip().lower() not in ("dual", "lgtm"): + return [] + try: + from temporalio.contrib.opentelemetry import TracingInterceptor + except ImportError: + return [] + return [TracingInterceptor()] + + async def get_temporal_client( temporal_address: str, metrics_url: str | None = None, @@ -136,6 +154,10 @@ async def get_temporal_client( dc = dataclasses.replace(dc, payload_codec=payload_codec) connect_kwargs["data_converter"] = dc + otel_interceptors = _build_otel_interceptors() + if otel_interceptors: + connect_kwargs["interceptors"] = otel_interceptors + if not metrics_url: client = await Client.connect(**connect_kwargs) else: