chore(tracing): TEMP traceparent probes across ACP->Temporal chain

Debug-only: log the active W3C traceparent at every hop (acp.handle_jsonrpc ->
before_create_task -> process_request bg-task -> svc.send_event/submit_task ->
client.send_signal/start_workflow [+ live interceptor list] -> worker.activity.
start_span) to pinpoint where async trace context is dropped. Grep "[TP-DEBUG]".
Remove before merge (rg TP-DEBUG / delete _tp_debug.py).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: NiteshDhanpal

src/agentex/lib/core/clients/temporal/temporal_client.py

@@ -10,6 +10,7 @@
 from temporalio.converter import PayloadCodec, DataConverter
 
 from agentex.lib.utils.logging import make_logger
+from agentex.lib.utils._tp_debug import log_tp  # TEMP(obs-debug)
 from agentex.lib.utils.model_utils import BaseModel
 from agentex.lib.core.clients.temporal.types import (
     TaskStatus,
@@ -157,6 +158,11 @@ async def start_workflow(
         **kwargs: Any,
     ) -> str:
         temporal_retry_policy = TemporalRetryPolicy(**retry_policy.model_dump(exclude_unset=True))
+        try:
+            _ics = [type(i).__name__ for i in self.client.config().get("interceptors", [])]
+        except Exception as _e:  # pragma: no cover - debug only
+            _ics = [f"<err:{_e}>"]
+        log_tp("client.start_workflow(before start)", interceptors=_ics)  # TEMP(obs-debug)
         workflow_handle = await self.client.start_workflow(
             *args,
             retry_policy=temporal_retry_policy,
@@ -174,6 +180,12 @@ async def send_signal(
         payload: dict[str, Any] | list[Any] | str | int | float | bool | BaseModel,
     ) -> None:
         handle = self.client.get_workflow_handle(workflow_id=workflow_id)
+        # TEMP(obs-debug): also report how many interceptors the live client has
+        try:
+            _ics = [type(i).__name__ for i in self.client.config().get("interceptors", [])]
+        except Exception as _e:  # pragma: no cover - debug only
+            _ics = [f"<err:{_e}>"]
+        log_tp("client.send_signal(before handle.signal)", wf=workflow_id, interceptors=_ics)  # TEMP(obs-debug)
         await handle.signal(signal, payload)  # type: ignore[misc]
 
     async def query_workflow(

src/agentex/lib/core/temporal/services/temporal_task_service.py

@@ -8,6 +8,7 @@
 from agentex.types.agent import Agent
 from agentex.types.event import Event
 from agentex.protocol.acp import SendEventParams, CreateTaskParams, InterruptTaskParams
+from agentex.lib.utils._tp_debug import log_tp  # TEMP(obs-debug)
 from agentex.lib.environment_variables import EnvironmentVariables
 from agentex.lib.core.clients.temporal.types import WorkflowState
 from agentex.lib.core.temporal.types.workflow import SignalName
@@ -67,6 +68,7 @@ async def submit_task(self, agent: Agent, task: Task, params: dict[str, Any] | N
         timeout_seconds = self._env_vars.WORKFLOW_EXECUTION_TIMEOUT_SECONDS
         execution_timeout = timedelta(seconds=timeout_seconds) if timeout_seconds and timeout_seconds > 0 else None
         with _acp_dispatch_span(f"acp.task_create:{task.id}"):
+            log_tp("svc.submit_task(before start_workflow)", task=task.id)  # TEMP(obs-debug)
             return await self._temporal_client.start_workflow(
                 workflow=self._env_vars.WORKFLOW_NAME,
                 arg=CreateTaskParams(
@@ -89,6 +91,7 @@ async def get_state(self, task_id: str) -> WorkflowState:
 
     async def send_event(self, agent: Agent, task: Task, event: Event, request: dict | None = None) -> None:
         with _acp_dispatch_span(f"acp.event_send:{task.id}"):
+            log_tp("svc.send_event(before send_signal)", task=task.id)  # TEMP(obs-debug)
             return await self._temporal_client.send_signal(
                 workflow_id=task.id,
                 signal=SignalName.RECEIVE_EVENT.value,

src/agentex/lib/core/tracing/trace.py

@@ -10,6 +10,7 @@
 from agentex import Agentex, AsyncAgentex
 from agentex.types.span import Span
 from agentex.lib.utils.logging import make_logger
+from agentex.lib.utils._tp_debug import log_tp  # TEMP(obs-debug)
 from agentex.lib.utils.model_utils import recursive_model_dump
 from agentex.lib.core.tracing.obs_ids import obs_correlation
 from agentex.lib.core.tracing.obs_span import (
@@ -151,6 +152,7 @@ def start_span(
         # _in_temporal_activity().
         id = str(uuid.uuid4())
         if _in_temporal_activity():
+            log_tp("worker.activity.start_span", span=name, biz_trace=self.trace_id)  # TEMP(obs-debug)
             tag_ambient_obs_span(business_span_id=id, business_trace_id=self.trace_id)
             obs_handle = None
             obs = obs_correlation()
@@ -331,6 +333,7 @@ async def start_span(
         # _in_temporal_activity().
         id = str(uuid.uuid4())
         if _in_temporal_activity():
+            log_tp("worker.activity.start_span", span=name, biz_trace=self.trace_id)  # TEMP(obs-debug)
             tag_ambient_obs_span(business_span_id=id, business_trace_id=self.trace_id)
             obs_handle = None
             obs = obs_correlation()

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

@@ -26,6 +26,7 @@
 )
 from agentex.lib.utils.logging import make_logger, ctx_var_request_id
 from agentex.protocol.json_rpc import JSONRPCError, JSONRPCRequest, JSONRPCResponse
+from agentex.lib.utils._tp_debug import log_tp  # TEMP(obs-debug)
 from agentex.lib.utils.model_utils import BaseModel
 from agentex.lib.utils.registration import register_agent
 
@@ -197,6 +198,8 @@ async def _handle_jsonrpc(self, request: Request):
                 params_data["request"] = {"headers": custom_headers}
             params = params_model.model_validate(params_data)
 
+            log_tp("acp.handle_jsonrpc", method=method, inbound=request.headers.get("traceparent"))  # TEMP(obs-debug)
+
             if method in RPC_SYNC_METHODS:
                 handler = self._handlers[method]
                 result = await handler(params)
@@ -223,6 +226,7 @@ async def _handle_jsonrpc(self, request: Request):
                     return JSONRPCResponse(id=None)
 
                 # For regular requests, start processing in background but return immediately
+                log_tp("acp.before_create_task", method=method)  # TEMP(obs-debug)
                 asyncio.create_task(
                     self._process_request(rpc_request.id, method, params)
                 )
@@ -304,6 +308,7 @@ async def _process_request(
     ):
         """Process a request in the background"""
         try:
+            log_tp("acp.process_request(bg-task)", method=method)  # TEMP(obs-debug)
             handler = self._handlers[method]
             await handler(params)
             # Note: In a real implementation, you might want to store the result somewhere

src/agentex/lib/utils/_tp_debug.py

@@ -0,0 +1,28 @@
+# TEMP(obs-debug): remove after diagnosing async-trace propagation.
+# Logs the W3C traceparent derived from the CURRENT active OpenTelemetry context
+# at each hop of the ACP ingress -> Temporal start/signal -> workflow -> activity
+# chain, so we can see exactly where the trace context is dropped. Grep the agent
+# logs for "[TP-DEBUG]".
+from __future__ import annotations
+
+from agentex.lib.utils.logging import make_logger
+
+logger = make_logger("tp_debug")
+
+
+def active_traceparent() -> str:
+    """W3C traceparent for the current active OTel context, or a marker string."""
+    try:
+        from opentelemetry.propagate import inject
+
+        carrier: dict[str, str] = {}
+        inject(carrier)
+        return carrier.get("traceparent", "<none>")
+    except Exception as exc:  # pragma: no cover - debug only
+        return f"<err:{exc}>"
+
+
+def log_tp(where: str, **extra: object) -> None:
+    """Emit a one-line [TP-DEBUG] log: where + current traceparent + any extras."""
+    kv = " ".join(f"{k}={v}" for k, v in extra.items())
+    logger.info("[TP-DEBUG] %s traceparent=%s %s", where, active_traceparent(), kv)