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)