fix(tracing): continue inbound W3C trace context at the ACP boundary

src/agentex/lib/sdk/fastacp/base/base_acp_server.py

@@ -46,19 +46,76 @@
 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) 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.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
+
+
+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."""
 
     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):

tests/test_trace_context_extraction.py

@@ -0,0 +1,87 @@
+"""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_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)