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: