Skip to content

Commit f831957

Browse files
NiteshDhanpalclaude
andcommitted
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-<trace>...) but FastACP never extracts it, so the app's active OTel context is <none>. 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 <noreply@anthropic.com>
1 parent 2a9576b commit f831957

2 files changed

Lines changed: 97 additions & 2 deletions

File tree

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

Lines changed: 47 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -46,19 +46,64 @@
4646
task_message_update_adapter = TypeAdapter(TaskMessageUpdate)
4747

4848

49+
def _attach_incoming_otel_context(scope_headers: list[tuple[bytes, bytes]]) -> object | None:
50+
"""Extract the inbound W3C trace context (traceparent/tracestate/baggage) from
51+
ASGI headers and make it the active OpenTelemetry context for the request.
52+
53+
FastACP is not otherwise instrumented to *continue* an incoming trace: the
54+
gateway forwards the traceparent header, but nothing on the Python side
55+
extracts it, so the active context stays empty. Downstream that means the
56+
Temporal ``start_workflow`` / ``signal`` (including the work dispatched via
57+
``asyncio.create_task``) fires with no active span, the interceptor injects
58+
nothing, and the workflow + activities detach into fresh traces.
59+
60+
Attaching here (in the ASGI middleware that wraps the whole request) fixes
61+
that: the request handler and the background task both run under the ingress
62+
trace, so the interceptor propagates it across the Temporal boundary.
63+
Returns a detach token (or None); fail-open.
64+
"""
65+
try:
66+
from opentelemetry import context as _otel_context
67+
from opentelemetry.propagate import extract
68+
69+
carrier = {k.decode("latin-1"): v.decode("latin-1") for k, v in scope_headers}
70+
return _otel_context.attach(extract(carrier))
71+
except Exception: # pragma: no cover - obs must never break a request
72+
return None
73+
74+
75+
def _detach_otel_context(token: object | None) -> None:
76+
if token is None:
77+
return
78+
try:
79+
from opentelemetry import context as _otel_context
80+
81+
_otel_context.detach(token) # type: ignore[arg-type]
82+
except Exception: # pragma: no cover - best-effort
83+
pass
84+
85+
4986
class RequestIDMiddleware:
5087
"""Pure ASGI middleware to set request IDs without buffering streaming responses."""
5188

5289
def __init__(self, app: ASGIApp) -> None:
5390
self.app = app
5491

5592
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
93+
otel_token: object | None = None
5694
if scope["type"] == "http":
57-
headers = dict(scope.get("headers", []))
95+
scope_headers = scope.get("headers", [])
96+
headers = dict(scope_headers)
5897
raw_request_id = headers.get(b"x-request-id", b"")
5998
request_id = raw_request_id.decode() if raw_request_id else uuid.uuid4().hex
6099
ctx_var_request_id.set(request_id)
61-
await self.app(scope, receive, send)
100+
# Continue the ingress trace for this request (and its background
101+
# Temporal dispatch); see _attach_incoming_otel_context.
102+
otel_token = _attach_incoming_otel_context(scope_headers)
103+
try:
104+
await self.app(scope, receive, send)
105+
finally:
106+
_detach_otel_context(otel_token)
62107

63108

64109
class BaseACPServer(FastAPI):
Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
"""Unit tests for ACP inbound W3C trace-context extraction.
2+
3+
Regression guard for the async end-to-end tracing fix: FastACP must *continue*
4+
an incoming traceparent (make it the active OpenTelemetry context) so the
5+
downstream Temporal start/signal — and the work dispatched via
6+
asyncio.create_task — run under the ingress trace instead of detaching into a
7+
fresh trace. See RequestIDMiddleware / _attach_incoming_otel_context.
8+
"""
9+
10+
from __future__ import annotations
11+
12+
from opentelemetry.propagate import inject
13+
14+
from agentex.lib.sdk.fastacp.base.base_acp_server import (
15+
_detach_otel_context,
16+
_attach_incoming_otel_context,
17+
)
18+
19+
20+
def _active_traceparent() -> str | None:
21+
carrier: dict[str, str] = {}
22+
inject(carrier)
23+
return carrier.get("traceparent")
24+
25+
26+
def test_attach_makes_inbound_traceparent_the_active_context() -> None:
27+
trace_id = "0af7651916cd43dd8448eb211c80319c"
28+
headers = [
29+
(b"traceparent", f"00-{trace_id}-b7ad6b7169203331-01".encode()),
30+
(b"content-type", b"application/json"),
31+
]
32+
token = _attach_incoming_otel_context(headers)
33+
try:
34+
active = _active_traceparent()
35+
assert active is not None, "no active traceparent after attach"
36+
# The active context must carry the ingress trace id, so the Temporal
37+
# interceptor propagates it downstream instead of starting a fresh trace.
38+
assert trace_id in active, f"expected ingress trace {trace_id}, got {active}"
39+
finally:
40+
_detach_otel_context(token)
41+
42+
43+
def test_no_inbound_traceparent_is_fail_open() -> None:
44+
# No traceparent header: must not raise, and detach must be safe.
45+
token = _attach_incoming_otel_context([(b"content-type", b"application/json")])
46+
_detach_otel_context(token)
47+
48+
49+
def test_detach_none_is_safe() -> None:
50+
_detach_otel_context(None)

0 commit comments

Comments
 (0)