Skip to content

feat(tracing): establish W3C trace context at FastACP ingress #466

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
NiteshDhanpal wants to merge 1 commit into
base: obs/01-span-obs-correlation
Choose a base branch
from
+52 −6
Open

feat(tracing): establish W3C trace context at FastACP ingress #466

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 52 additions & 6 deletions src/agentex/lib/sdk/fastacp/base/base_acp_server.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,19 +46,65 @@
task_message_update_adapter = TypeAdapter(TaskMessageUpdate)


def _attach_incoming_trace_context(scope: Scope) -> Any:
"""Extract the W3C trace context (traceparent/tracestate) from the inbound
request headers and attach it as the current OpenTelemetry context, so spans
created while handling this request adopt the caller's observability trace.

Returns a detach token (pass to opentelemetry.context.detach) or None if OTel
is unavailable or no context could be established. Best-effort: never raises.
"""
try:
from opentelemetry import context as otel_context
from opentelemetry.propagate import extract
except ImportError:
return None
try:
carrier = {
k.decode("latin-1"): v.decode("latin-1")
for k, v in scope.get("headers", [])
}
ctx = extract(carrier)
Comment on lines +62 to +67

@greptile-apps greptile-apps Bot Jul 21, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Duplicate tracestate headers silently dropped

The plain-dict carrier overwrites duplicate keys, so if an intermediary forwards multiple tracestate headers (which the W3C Trace Context spec explicitly permits), all but the last entry are silently discarded. The OTel DefaultGetter calls carrier.get(key) on a plain dict, which can only return a single string value, so combined trace-state from several systems in the propagation chain would be partially lost.

To handle multi-value headers correctly, the carrier should concatenate repeated values with a comma separator before inserting them into the dict, or use a multi-value mapping with a custom Getter.

Prompt To Fix With AI 
This is a comment left during a code review.
Path: src/agentex/lib/sdk/fastacp/base/base_acp_server.py
Line: 62-67

Comment:
**Duplicate `tracestate` headers silently dropped**

The plain-dict carrier overwrites duplicate keys, so if an intermediary forwards multiple `tracestate` headers (which the W3C Trace Context spec explicitly permits), all but the last entry are silently discarded. The OTel `DefaultGetter` calls `carrier.get(key)` on a plain dict, which can only return a single string value, so combined trace-state from several systems in the propagation chain would be partially lost.

To handle multi-value headers correctly, the carrier should concatenate repeated values with a comma separator before inserting them into the dict, or use a multi-value mapping with a custom `Getter`.

How can I resolve this? If you propose a fix, please make it concise.

Fix in Cursor Fix in Claude Code Fix in Codex

return otel_context.attach(ctx)
except Exception: # pragma: no cover - propagation must never break a request
return None


def _detach_trace_context(token: Any) -> None:
if token is None:
return
try:
from opentelemetry import context as otel_context

otel_context.detach(token)
except Exception: # pragma: no cover
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:
if scope["type"] == "http":
headers = dict(scope.get("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)
if scope["type"] != "http":
await self.app(scope, receive, send)
return

headers = dict(scope.get("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)

# Establish the caller's W3C trace context for the duration of the
# request so business spans created here (and any downstream Temporal
# workflow started from this request) share the same observability trace.
token = _attach_incoming_trace_context(scope)
try:
await self.app(scope, receive, send)
finally:
_detach_trace_context(token)


class BaseACPServer(FastAPI):
Expand Down
Loading