feat: propagate trace context and log correlation (#397)

Author: liujuanjuan1984

docs/guide.md

@@ -37,6 +37,9 @@ Key variables to understand protocol behavior:
 - `A2A_PUBLIC_URL`: public base URL advertised by the Agent Card. Default: `http://127.0.0.1:8000`.
 - `A2A_LOG_LEVEL`: runtime log level. Default: `WARNING`.
 - `A2A_LOG_PAYLOADS` / `A2A_LOG_BODY_LIMIT`: payload logging behavior and truncation. When `A2A_LOG_LEVEL=DEBUG`, upstream OpenCode stream events are also logged with preview truncation controlled by `A2A_LOG_BODY_LIMIT`.
+- The runtime accepts W3C `traceparent` / `tracestate` headers on inbound requests. When `traceparent` is missing or invalid, the runtime generates a fresh valid value and exposes it on the HTTP response header.
+- The active `traceparent` / `tracestate` pair is propagated across inbound A2A handling, OpenCode upstream requests, and outbound peer A2A calls triggered through the embedded client or `a2a_call` tool path.
+- Logs derive a stable `trace_id` from the active `traceparent` so request-scoped log lines can be correlated without introducing high-cardinality metric labels.
 - `A2A_HTTP_GZIP_MINIMUM_SIZE`: minimum eligible response-body size in bytes for global non-streaming HTTP gzip compression. Default: `8192`.
 - `A2A_MAX_REQUEST_BODY_BYTES`: runtime request-body limit. Oversized requests return HTTP `413`.
 - `A2A_PENDING_SESSION_CLAIM_TTL_SECONDS`: lease duration for pending preferred session claims before they expire and stop blocking other identities.

src/opencode_a2a/client/agent_card.py

@@ -13,6 +13,7 @@
     PREV_AGENT_CARD_WELL_KNOWN_PATH,
 )
 
+from ..trace_context import current_trace_headers
 from .request_context import build_default_headers
 
 
@@ -70,6 +71,9 @@ def build_resolver_http_kwargs(
 ) -> dict[str, Any]:
     http_kwargs: dict[str, Any] = {"timeout": timeout}
     default_headers = build_default_headers(bearer_token, basic_auth)
+    trace_headers = current_trace_headers()
+    if trace_headers:
+        default_headers.update(trace_headers)
     if default_headers:
         http_kwargs["headers"] = default_headers
     return http_kwargs

src/opencode_a2a/client/client.py

@@ -35,7 +35,7 @@
 from .errors import A2ATimeoutError, A2AUnsupportedBindingError
 from .payload_text import extract_text as extract_text_from_payload
 from .polling import PollingFallbackPolicy
-from .request_context import build_call_context, build_client_interceptors, split_request_metadata
+from .request_context import build_call_context, split_request_metadata
 
 
 class A2AClient:
@@ -302,14 +302,7 @@ async def _build_client(self) -> Client:
         )
         try:
             factory = ClientFactory(config, consumers=None)
-            client = factory.create(
-                card,
-                interceptors=build_client_interceptors(
-                    self._settings.bearer_token,
-                    self._settings.basic_auth,
-                    self._settings.protocol_version,
-                ),
-            )
+            client = factory.create(card)
         except ValueError as exc:
             raise A2AUnsupportedBindingError(
                 f"No supported transport found for {self.agent_url}"

src/opencode_a2a/client/request_context.py

@@ -5,40 +5,13 @@
 from collections.abc import Mapping
 from typing import Any
 
-from a2a.client.middleware import ClientCallContext, ClientCallInterceptor
+from a2a.client.middleware import ClientCallContext
 
 from ..protocol_versions import normalize_protocol_version
+from ..trace_context import current_trace_headers
 from .auth import encode_basic_auth
 
 
-class HeaderInterceptor(ClientCallInterceptor):
-    def __init__(self, default_headers: Mapping[str, str] | None = None) -> None:
-        self._default_headers = {
-            key: value for key, value in dict(default_headers or {}).items() if value is not None
-        }
-
-    async def intercept(
-        self,
-        method_name: str,
-        request_payload: dict[str, Any],
-        http_kwargs: dict[str, Any],
-        agent_card: object | None,
-        context: ClientCallContext | None,
-    ) -> tuple[dict[str, Any], dict[str, Any]]:
-        del method_name, agent_card
-        headers = dict(http_kwargs.get("headers") or {})
-        headers.update(self._default_headers)
-        if context is not None:
-            dynamic_headers = context.state.get("headers")
-            if isinstance(dynamic_headers, Mapping):
-                for key, value in dynamic_headers.items():
-                    if isinstance(key, str) and value is not None:
-                        headers[key] = str(value)
-        if headers:
-            http_kwargs["headers"] = headers
-        return request_payload, http_kwargs
-
-
 def build_default_headers(
     bearer_token: str | None,
     basic_auth: str | None = None,
@@ -68,6 +41,14 @@ def split_request_metadata(
             if value is not None:
                 extra_headers["A2A-Version"] = normalize_protocol_version(str(value))
             continue
+        if isinstance(key, str) and key.lower() == "traceparent":
+            if value is not None:
+                extra_headers["traceparent"] = str(value)
+            continue
+        if isinstance(key, str) and key.lower() == "tracestate":
+            if value is not None:
+                extra_headers["tracestate"] = str(value)
+            continue
         request_metadata[key] = value
     return request_metadata or None, extra_headers or None
 
@@ -79,6 +60,7 @@ def build_call_context(
     protocol_version: str | None = None,
 ) -> ClientCallContext | None:
     merged_headers = build_default_headers(bearer_token, basic_auth, protocol_version)
+    merged_headers.update(current_trace_headers())
     if extra_headers:
         merged_headers.update(extra_headers)
     if not merged_headers:
@@ -91,18 +73,8 @@ def build_call_context(
     )
 
 
-def build_client_interceptors(
-    bearer_token: str | None,
-    basic_auth: str | None = None,
-    protocol_version: str | None = None,
-) -> list[ClientCallInterceptor]:
-    return [HeaderInterceptor(build_default_headers(bearer_token, basic_auth, protocol_version))]
-
-
 __all__ = [
-    "HeaderInterceptor",
     "build_call_context",
-    "build_client_interceptors",
     "build_default_headers",
     "split_request_metadata",
 ]

src/opencode_a2a/execution/executor.py

@@ -214,6 +214,7 @@ async def execute(self, context: RequestContext, event_queue: EventQueue) -> Non
         identity = (call_context.state.get("identity") if call_context else None) or "anonymous"
         credential_id = call_context.state.get("credential_id") if call_context else None
         auth_scheme = call_context.state.get("auth_scheme") if call_context else None
+        trace_id = call_context.state.get("trace_id") if call_context else None
 
         streaming_request = self._should_stream(context)
         accepted_output_modes = normalize_accepted_output_modes(context.configuration)
@@ -313,13 +314,14 @@ async def execute(self, context: RequestContext, event_queue: EventQueue) -> Non
 
         logger.debug(
             (
-                "Received message identity=%s credential_id=%s auth_scheme=%s "
+                "Received message identity=%s credential_id=%s auth_scheme=%s trace_id=%s "
                 "task_id=%s context_id=%s "
                 "streaming=%s text=%s part_count=%s"
             ),
             identity,
             credential_id,
             auth_scheme,
+            trace_id,
             task_id,
             context_id,
             streaming_request,

src/opencode_a2a/opencode_upstream_client.py

@@ -15,6 +15,7 @@
 from .parts.text import extract_text_from_parts
 from .runtime_state import InterruptRequestBinding
 from .server.state_store import InterruptRequestRepository, MemoryInterruptRequestRepository
+from .trace_context import current_trace_headers
 
 _UNSET = object()
 logger = logging.getLogger(__name__)
@@ -143,6 +144,13 @@ def _build_http_client(self, base_url: str) -> httpx.AsyncClient:
             headers={"Accept": "application/json"},
         )
 
+    @staticmethod
+    def _request_headers(headers: Mapping[str, str] | None = None) -> dict[str, str] | None:
+        merged_headers = current_trace_headers()
+        if headers:
+            merged_headers.update(dict(headers))
+        return merged_headers or None
+
     async def close(self) -> None:
         await self._client.aclose()
 
@@ -185,7 +193,11 @@ async def _get_json(
         params: Mapping[str, Any] | None = None,
     ) -> Any:
         async with self._request_budget.reserve(operation=endpoint):
-            response = await self._client.get(path, params=params)
+            response = await self._client.get(
+                path,
+                params=params,
+                headers=self._request_headers(),
+            )
             response.raise_for_status()
             return self._decode_json_response(response, endpoint=endpoint)
 
@@ -207,6 +219,7 @@ async def _post_json(
             response = await self._client.post(
                 path,
                 params=params,
+                headers=self._request_headers(),
                 **request_kwargs,
             )
             response.raise_for_status()
@@ -238,7 +251,11 @@ async def _delete_json(
         params: Mapping[str, Any] | None = None,
     ) -> Any:
         async with self._request_budget.reserve(operation=endpoint):
-            response = await self._client.delete(path, params=params)
+            response = await self._client.delete(
+                path,
+                params=params,
+                headers=self._request_headers(),
+            )
             response.raise_for_status()
             return self._decode_json_response(response, endpoint=endpoint)
 
@@ -402,7 +419,7 @@ async def stream_events(
                 "/event",
                 params=params,
                 timeout=None,
-                headers={"Accept": "text/event-stream"},
+                headers=self._request_headers({"Accept": "text/event-stream"}),
             ) as response:
                 response.raise_for_status()
                 data_lines: list[str] = []
@@ -555,6 +572,7 @@ async def list_messages(
             response = await self._client.get(
                 f"/session/{session_id}/message",
                 params=self._merge_params(params, workspace_id=workspace_id),
+                headers=self._request_headers(),
             )
             response.raise_for_status()
             payload = self._decode_json_response(response, endpoint=endpoint)
@@ -594,6 +612,7 @@ async def session_prompt_async(
                 f"/session/{session_id}/prompt_async",
                 params=self._query_params(directory=directory, workspace_id=workspace_id),
                 json=request,
+                headers=self._request_headers(),
             )
             response.raise_for_status()
             if response.status_code != 204:
@@ -748,7 +767,10 @@ async def create_workspace(self, request: dict[str, Any]) -> Any:
 
     async def remove_workspace(self, workspace_id: str) -> Any:
         async with self._request_budget.reserve(operation="/experimental/workspace/{id}"):
-            response = await self._client.delete(f"/experimental/workspace/{workspace_id}")
+            response = await self._client.delete(
+                f"/experimental/workspace/{workspace_id}",
+                headers=self._request_headers(),
+            )
             response.raise_for_status()
             return self._decode_json_response(
                 response,
@@ -774,6 +796,7 @@ async def remove_worktree(self, request: dict[str, Any]) -> bool:
                 "DELETE",
                 "/experimental/worktree",
                 json=request,
+                headers=self._request_headers(),
             )
             response.raise_for_status()
             payload = self._decode_json_response(response, endpoint="/experimental/worktree")

src/opencode_a2a/server/application.py

@@ -66,6 +66,7 @@
 from ..opencode_upstream_client import OpencodeUpstreamClient
 from ..output_modes import normalize_accepted_output_modes
 from ..profile.runtime import build_runtime_profile
+from ..trace_context import install_log_record_factory
 from .agent_card import (
     _CHAT_OUTPUT_MODES,
     _build_agent_card_description,
@@ -514,6 +515,15 @@ def build(self, request: Request) -> ServerCallContext:
         credential_id = getattr(request.state, "user_credential_id", None)
         if credential_id:
             context.state["credential_id"] = credential_id
+        traceparent = getattr(request.state, "traceparent", None)
+        if traceparent:
+            context.state["traceparent"] = traceparent
+        tracestate = getattr(request.state, "tracestate", None)
+        if tracestate:
+            context.state["tracestate"] = tracestate
+        trace_id = getattr(request.state, "trace_id", None)
+        if trace_id:
+            context.state["trace_id"] = trace_id
         negotiated_protocol_version = getattr(request.state, "a2a_protocol_version", None)
         if negotiated_protocol_version:
             context.state["a2a_protocol_version"] = negotiated_protocol_version
@@ -525,6 +535,7 @@ def build(self, request: Request) -> ServerCallContext:
 
 
 def create_app(settings: Settings) -> FastAPI:
+    install_log_record_factory()
     database_engine = (
         build_database_engine(settings) if settings.a2a_task_store_backend == "database" else None
     )
@@ -663,9 +674,10 @@ def _normalize_log_level(value: str) -> str:
 
 
 def _configure_logging(level: str) -> None:
+    install_log_record_factory()
     logging.basicConfig(
         level=getattr(logging, level, logging.INFO),
-        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+        format="%(asctime)s %(levelname)s %(name)s [trace_id=%(trace_id)s]: %(message)s",
     )
     logging.getLogger("uvicorn.error").setLevel(level)
     logging.getLogger("uvicorn.access").setLevel(level)

src/opencode_a2a/server/middleware.py

@@ -30,6 +30,13 @@
     negotiate_protocol_version,
     normalize_protocol_version,
 )
+from ..trace_context import (
+    TRACEPARENT_HEADER,
+    TRACESTATE_HEADER,
+    reset_current_trace_context,
+    resolve_trace_context,
+    set_current_trace_context,
+)
 from .request_parsing import (
     _decode_payload_preview,
     _detect_sensitive_extension_method,
@@ -158,6 +165,24 @@ def _uses_v1_jsonrpc_aliases(request: Request) -> bool:
         except ValueError:
             return False
 
+    @app.middleware("http")
+    async def bind_trace_context(request: Request, call_next):
+        trace_context = resolve_trace_context(
+            request.headers.get(TRACEPARENT_HEADER),
+            request.headers.get(TRACESTATE_HEADER),
+        )
+        request.state.traceparent = trace_context.traceparent
+        request.state.trace_id = trace_context.trace_id
+        if trace_context.tracestate:
+            request.state.tracestate = trace_context.tracestate
+        token = set_current_trace_context(trace_context)
+        try:
+            response = await call_next(request)
+        finally:
+            reset_current_trace_context(token)
+        response.headers[TRACEPARENT_HEADER] = trace_context.traceparent
+        return response
+
     @app.middleware("http")
     async def negotiate_a2a_protocol_version(request: Request, call_next):
         token: Token | None = None