From ddb745dd2cc3f9733a4ba0f28a0bbafcdaa1a7a0 Mon Sep 17 00:00:00 2001 From: Nitesh Dhanpal Date: Wed, 5 Aug 2026 15:50:16 -0700 Subject: [PATCH 1/2] fix(tracing): continue inbound W3C trace context at the ACP boundary Root cause of async trace detachment (proven via [TP-DEBUG] probes): the ingress traceparent arrives in the HTTP header (inbound=00-...) but FastACP never extracts it, so the app's active OTel context is . Downstream the Temporal start_workflow/signal (incl. the asyncio.create_task background dispatch) fires with no active span, the interceptor injects nothing, and the workflow + every activity start FRESH traces disconnected from the ingress. Extract + attach the inbound W3C context in the ASGI RequestIDMiddleware (wraps the whole request, so the bg task inherits it via create_task's context copy). Now the interceptor propagates the ingress trace across the Temporal boundary and the workflow/activity inherit it -> one connected trace. Fail-open. Unit-tested. Co-Authored-By: Claude Opus 4.8 --- .../lib/sdk/fastacp/base/base_acp_server.py | 49 +++++++++++++++++- tests/test_trace_context_extraction.py | 50 +++++++++++++++++++ 2 files changed, 97 insertions(+), 2 deletions(-) create mode 100644 tests/test_trace_context_extraction.py diff --git a/src/agentex/lib/sdk/fastacp/base/base_acp_server.py b/src/agentex/lib/sdk/fastacp/base/base_acp_server.py index 1ea8e82e6..481ccdaa0 100644 --- a/src/agentex/lib/sdk/fastacp/base/base_acp_server.py +++ b/src/agentex/lib/sdk/fastacp/base/base_acp_server.py @@ -46,6 +46,43 @@ task_message_update_adapter = TypeAdapter(TaskMessageUpdate) +def _attach_incoming_otel_context(scope_headers: list[tuple[bytes, bytes]]) -> object | None: + """Extract the inbound W3C trace context (traceparent/tracestate/baggage) from + ASGI headers and make it the active OpenTelemetry context for the request. + + FastACP is not otherwise instrumented to *continue* an incoming trace: the + gateway forwards the traceparent header, but nothing on the Python side + extracts it, so the active context stays empty. Downstream that means the + Temporal ``start_workflow`` / ``signal`` (including the work dispatched via + ``asyncio.create_task``) fires with no active span, the interceptor injects + nothing, and the workflow + activities detach into fresh traces. + + Attaching here (in the ASGI middleware that wraps the whole request) fixes + that: the request handler and the background task both run under the ingress + trace, so the interceptor propagates it across the Temporal boundary. + Returns a detach token (or None); fail-open. + """ + try: + from opentelemetry import context as _otel_context + from opentelemetry.propagate import extract + + carrier = {k.decode("latin-1"): v.decode("latin-1") for k, v in scope_headers} + return _otel_context.attach(extract(carrier)) + except Exception: # pragma: no cover - obs must never break a request + return None + + +def _detach_otel_context(token: object | None) -> None: + if token is None: + return + try: + from opentelemetry import context as _otel_context + + _otel_context.detach(token) # type: ignore[arg-type] + except Exception: # pragma: no cover - best-effort + pass + + class RequestIDMiddleware: """Pure ASGI middleware to set request IDs without buffering streaming responses.""" @@ -53,12 +90,20 @@ def __init__(self, app: ASGIApp) -> None: self.app = app async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + otel_token: object | None = None if scope["type"] == "http": - headers = dict(scope.get("headers", [])) + scope_headers = scope.get("headers", []) + headers = dict(scope_headers) raw_request_id = headers.get(b"x-request-id", b"") request_id = raw_request_id.decode() if raw_request_id else uuid.uuid4().hex ctx_var_request_id.set(request_id) - await self.app(scope, receive, send) + # Continue the ingress trace for this request (and its background + # Temporal dispatch); see _attach_incoming_otel_context. + otel_token = _attach_incoming_otel_context(scope_headers) + try: + await self.app(scope, receive, send) + finally: + _detach_otel_context(otel_token) class BaseACPServer(FastAPI): diff --git a/tests/test_trace_context_extraction.py b/tests/test_trace_context_extraction.py new file mode 100644 index 000000000..03e523c15 --- /dev/null +++ b/tests/test_trace_context_extraction.py @@ -0,0 +1,50 @@ +"""Unit tests for ACP inbound W3C trace-context extraction. + +Regression guard for the async end-to-end tracing fix: FastACP must *continue* +an incoming traceparent (make it the active OpenTelemetry context) so the +downstream Temporal start/signal — and the work dispatched via +asyncio.create_task — run under the ingress trace instead of detaching into a +fresh trace. See RequestIDMiddleware / _attach_incoming_otel_context. +""" + +from __future__ import annotations + +from opentelemetry.propagate import inject + +from agentex.lib.sdk.fastacp.base.base_acp_server import ( + _detach_otel_context, + _attach_incoming_otel_context, +) + + +def _active_traceparent() -> str | None: + carrier: dict[str, str] = {} + inject(carrier) + return carrier.get("traceparent") + + +def test_attach_makes_inbound_traceparent_the_active_context() -> None: + trace_id = "0af7651916cd43dd8448eb211c80319c" + headers = [ + (b"traceparent", f"00-{trace_id}-b7ad6b7169203331-01".encode()), + (b"content-type", b"application/json"), + ] + token = _attach_incoming_otel_context(headers) + try: + active = _active_traceparent() + assert active is not None, "no active traceparent after attach" + # The active context must carry the ingress trace id, so the Temporal + # interceptor propagates it downstream instead of starting a fresh trace. + assert trace_id in active, f"expected ingress trace {trace_id}, got {active}" + finally: + _detach_otel_context(token) + + +def test_no_inbound_traceparent_is_fail_open() -> None: + # No traceparent header: must not raise, and detach must be safe. + token = _attach_incoming_otel_context([(b"content-type", b"application/json")]) + _detach_otel_context(token) + + +def test_detach_none_is_safe() -> None: + _detach_otel_context(None) From f13c941d7902cca92942f8f2d12cd133ba72277a Mon Sep 17 00:00:00 2001 From: Nitesh Dhanpal Date: Mon, 10 Aug 2026 22:04:32 -0700 Subject: [PATCH 2/2] fix(tracing): parse ACP ingress headers with explicit W3C propagator + repeated-header support - Use TraceContextTextMapPropagator().extract() instead of the ambient global propagator, so OTEL_PROPAGATORS=datadog (plausible in a DD shop; dd_only is the default mode) can't silently disable W3C extraction. - Build a dict-of-lists carrier so repeated `tracestate` header lines are combined (W3C/RFC7230 MUST) instead of collapsing to the last value. - Parse only traceparent/tracestate, so arbitrary inbound `baggage` isn't pulled into the downstream context. Addresses the review comments on repeated-header collapse, ambient-propagator dependence, and the implicit trust boundary. `traceparent` is single-valued, so trace linkage (this PR's purpose) is unchanged. Tests: repeated tracestate combined; inbound baggage not extracted. Co-Authored-By: Claude Opus 4.8 --- .../lib/sdk/fastacp/base/base_acp_server.py | 24 +++++++++--- tests/test_trace_context_extraction.py | 37 +++++++++++++++++++ 2 files changed, 55 insertions(+), 6 deletions(-) diff --git a/src/agentex/lib/sdk/fastacp/base/base_acp_server.py b/src/agentex/lib/sdk/fastacp/base/base_acp_server.py index 481ccdaa0..864b466d0 100644 --- a/src/agentex/lib/sdk/fastacp/base/base_acp_server.py +++ b/src/agentex/lib/sdk/fastacp/base/base_acp_server.py @@ -47,8 +47,8 @@ def _attach_incoming_otel_context(scope_headers: list[tuple[bytes, bytes]]) -> object | None: - """Extract the inbound W3C trace context (traceparent/tracestate/baggage) from - ASGI headers and make it the active OpenTelemetry context for the request. + """Extract the inbound W3C trace context (traceparent/tracestate) from ASGI + headers and make it the active OpenTelemetry context for the request. FastACP is not otherwise instrumented to *continue* an incoming trace: the gateway forwards the traceparent header, but nothing on the Python side @@ -64,10 +64,22 @@ def _attach_incoming_otel_context(scope_headers: list[tuple[bytes, bytes]]) -> o """ try: from opentelemetry import context as _otel_context - from opentelemetry.propagate import extract - - carrier = {k.decode("latin-1"): v.decode("latin-1") for k, v in scope_headers} - return _otel_context.attach(extract(carrier)) + from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator + + # ASGI headers are a list that can repeat a name, and a dict comprehension + # keeps only the last value -- which silently drops repeated `tracestate` + # lines (W3C/RFC7230 say they MUST be combined). Build a dict-of-lists so + # the propagator's getter sees every value and `TraceState.from_header` + # combines them; `traceparent` is single-valued so it is unaffected. + carrier: dict[str, list[str]] = {} + for k, v in scope_headers: + carrier.setdefault(k.decode("latin-1").lower(), []).append(v.decode("latin-1")) + # Use the W3C propagator explicitly rather than the ambient global one: + # this ingress is W3C by contract, and `OTEL_PROPAGATORS=datadog` (plausible + # in a DD shop, and dd_only is the default mode) would otherwise silently + # extract nothing. It also parses only traceparent/tracestate, so arbitrary + # inbound `baggage` is not pulled into the downstream context. + return _otel_context.attach(TraceContextTextMapPropagator().extract(carrier)) except Exception: # pragma: no cover - obs must never break a request return None diff --git a/tests/test_trace_context_extraction.py b/tests/test_trace_context_extraction.py index 03e523c15..3da6fd55d 100644 --- a/tests/test_trace_context_extraction.py +++ b/tests/test_trace_context_extraction.py @@ -46,5 +46,42 @@ def test_no_inbound_traceparent_is_fail_open() -> None: _detach_otel_context(token) +def test_repeated_tracestate_headers_are_combined() -> None: + # ASGI can deliver tracestate as multiple header lines; W3C/RFC7230 require + # combining them. The old dict-comprehension carrier kept only the last. + from opentelemetry import trace as _trace + + trace_id = "0af7651916cd43dd8448eb211c80319c" + headers = [ + (b"traceparent", f"00-{trace_id}-b7ad6b7169203331-01".encode()), + (b"tracestate", b"vendora=1"), + (b"tracestate", b"vendorb=2"), + ] + token = _attach_incoming_otel_context(headers) + try: + ts = _trace.get_current_span().get_span_context().trace_state + assert ts.get("vendora") == "1" + assert ts.get("vendorb") == "2" # would be missing if repeats collapsed + finally: + _detach_otel_context(token) + + +def test_inbound_baggage_is_not_extracted() -> None: + # W3C tracecontext-only extraction: arbitrary inbound baggage (attacker- + # controlled keys) must not be pulled into the downstream context. + from opentelemetry.baggage import get_all + + trace_id = "0af7651916cd43dd8448eb211c80319c" + headers = [ + (b"traceparent", f"00-{trace_id}-b7ad6b7169203331-01".encode()), + (b"baggage", b"user_id=secret,role=admin"), + ] + token = _attach_incoming_otel_context(headers) + try: + assert get_all() == {} + finally: + _detach_otel_context(token) + + def test_detach_none_is_safe() -> None: _detach_otel_context(None)