feat(streaming): tag streamed events with agent_path

Streamed lifecycle/text events carried only a task_id, so events from
multiple agents sharing one task stream (e.g. child-workflow sub-agents
emitting into the orchestrator's stream) were indistinguishable.

Thread an optional agent_path (str | list[str]) end to end:
- TaskMessage envelope carries it, so every start/delta/full/done event
  is attributable via parent_task_message.
- ContextInterceptor propagates it workflow->activity via a new
  ContextVar + header, so the model activity tags live text/reasoning deltas.
- TemporalStreamingHooks + stream_lifecycle_content + streaming service
  stamp it on tool/lifecycle events.

Backward compatible: agent_path defaults to None (untagged) everywhere.

NOTE: task_message.py is Stainless-generated; the agent_path field is a
manual stopgap and must be mirrored in the upstream OpenAPI spec
(.stats.yml openapi_spec_url) or the next regen will drop it.

Author: vkalmathscale

src/agentex/lib/adk/_modules/streaming.py

@@ -54,6 +54,7 @@ def streaming_task_message_context(
         initial_content: TaskMessageContent,
         streaming_mode: StreamingMode = "coalesced",
         created_at: datetime | None = None,
+        agent_path: str | list[str] | None = None,
     ) -> StreamingTaskMessageContext:
         """
         Create a streaming context for managing TaskMessage lifecycle.
@@ -86,4 +87,5 @@ def streaming_task_message_context(
             initial_content=initial_content,
             streaming_mode=streaming_mode,
             created_at=created_at,
+            agent_path=agent_path,
         )

src/agentex/lib/core/services/adk/streaming.py

@@ -379,10 +379,12 @@ def __init__(
         streaming_service: "StreamingService",
         streaming_mode: StreamingMode = "coalesced",
         created_at: datetime | None = None,
+        agent_path: str | list[str] | None = None,
     ):
         self.task_id = task_id
         self.initial_content = initial_content
         self.task_message: TaskMessage | None = None
+        self.agent_path = agent_path
         self._agentex_client = agentex_client
         self._streaming_service = streaming_service
         self._is_closed = False
@@ -406,6 +408,11 @@ async def open(self) -> "StreamingTaskMessageContext":
             streaming_status="IN_PROGRESS",
             created_at=self._created_at if self._created_at is not None else omit,
         )
+        # Stamped on the in-memory envelope only: every start/delta/full/done event
+        # carries it via parent_task_message. Not sent to messages.create/update —
+        # the persisted-message contract is unchanged; this is a stream-only tag.
+        if self.agent_path is not None:
+            self.task_message.agent_path = self.agent_path
 
         # Send the START event
         start_event = StreamTaskMessageStart(
@@ -540,6 +547,7 @@ def streaming_task_message_context(
         initial_content: TaskMessageContent,
         streaming_mode: StreamingMode = "coalesced",
         created_at: datetime | None = None,
+        agent_path: str | list[str] | None = None,
     ) -> StreamingTaskMessageContext:
         return StreamingTaskMessageContext(
             task_id=task_id,
@@ -548,6 +556,7 @@ def streaming_task_message_context(
             streaming_service=self,
             streaming_mode=streaming_mode,
             created_at=created_at,
+            agent_path=agent_path,
         )
 
     async def stream_update(self, update: TaskMessageUpdate) -> TaskMessageUpdate | None:

src/agentex/lib/core/temporal/plugins/openai_agents/hooks/activities.py

@@ -38,6 +38,7 @@ def _deserialize_content(data: Dict[str, Any]):
 async def stream_lifecycle_content(
     task_id: str,
     content: Dict[str, Any],
+    agent_path: str | list[str] | None = None,
 ) -> None:
     """Stream agent lifecycle content to the AgentEx UI.
 
@@ -58,6 +59,9 @@ async def stream_lifecycle_content(
             - type="text": TextContent (plain text messages, handoff notifications)
             - type="tool_request": ToolRequestContent (tool invocation with call_id)
             - type="tool_response": ToolResponseContent (tool result with call_id)
+        agent_path: Optional emitting-agent identifier stamped on the streamed
+            envelope so consumers can attribute events when multiple agents share
+            one task stream.
 
     Note:
         This activity is non-blocking and will not throw exceptions to the workflow.
@@ -69,6 +73,7 @@ async def stream_lifecycle_content(
         async with adk.streaming.streaming_task_message_context(
             task_id=task_id,
             initial_content=typed_content,
+            agent_path=agent_path,
         ) as streaming_context:
             await streaming_context.stream_update(
                 StreamTaskMessageFull(

src/agentex/lib/core/temporal/plugins/openai_agents/hooks/hooks.py

@@ -121,6 +121,7 @@ async def on_agent_start(self, context, agent):
         emit_handoffs: Whether to stream the handoff text message
         trace_id: When set, tool calls are traced to SGP (input + output)
         parent_span_id: Parent span for the per-tool spans
+        agent_path: Emitting-agent identifier stamped on streamed events
     """
 
     def __init__(
@@ -133,6 +134,7 @@ def __init__(
         emit_handoffs: bool = True,
         trace_id: str | None = None,
         parent_span_id: str | None = None,
+        agent_path: str | list[str] | None = None,
     ):
         """Initialize the streaming hooks.
 
@@ -158,6 +160,10 @@ def __init__(
                 the tool) with the arguments as input and the result as output. When None,
                 no tool spans are created (token-usage metrics still emit).
             parent_span_id: Parent span id the per-tool spans attach to.
+            agent_path: Identifier of the agent these hooks stream for — a single
+                id or a root->emitter path (e.g. ["researcher", "subagent-abc"]).
+                Stamped on every emitted event so consumers can attribute it when
+                multiple agents share one task stream. Defaults to None (untagged).
         """
         super().__init__()
         self.task_id = task_id
@@ -167,6 +173,7 @@ def __init__(
         self.emit_handoffs = emit_handoffs
         self.trace_id = trace_id
         self.parent_span_id = parent_span_id
+        self.agent_path = agent_path
         # tool_call_id -> open SGP span, so on_tool_end closes the right one.
         self._tool_spans: dict[str, Any] = {}
 
@@ -247,6 +254,7 @@ async def on_tool_start(self, context: RunContextWrapper, agent: Agent, tool: To
                         name=tool.name,
                         arguments=tool_arguments,
                     ).model_dump(),
+                    self.agent_path,
                 ],
                 start_to_close_timeout=self.timeout,
             )
@@ -286,6 +294,7 @@ async def on_tool_end(
                         name=tool.name,
                         content=result,
                     ).model_dump(),
+                    self.agent_path,
                 ],
                 start_to_close_timeout=self.timeout,
             )
@@ -320,6 +329,7 @@ async def on_handoff(
                     content=f"Handoff from {from_agent.name} to {to_agent.name}",
                     type="text",
                 ).model_dump(),
+                self.agent_path,
             ],
             start_to_close_timeout=self.timeout,
         )

src/agentex/lib/core/temporal/plugins/openai_agents/interceptors/context_interceptor.py

@@ -7,7 +7,7 @@
 """
 
 import logging
-from typing import Any, Type, Optional, override
+from typing import Any, List, Type, Union, Optional, override
 from contextvars import ContextVar
 
 from temporalio import workflow
@@ -30,11 +30,13 @@
 streaming_task_id: ContextVar[Optional[str]] = ContextVar('streaming_task_id', default=None)
 streaming_trace_id: ContextVar[Optional[str]] = ContextVar('streaming_trace_id', default=None)
 streaming_parent_span_id: ContextVar[Optional[str]] = ContextVar('streaming_parent_span_id', default=None)
+streaming_agent_path: ContextVar[Optional[Union[str, List[str]]]] = ContextVar('streaming_agent_path', default=None)
 
 # Header keys for passing context
 TASK_ID_HEADER = "context-task-id"
 TRACE_ID_HEADER = "context-trace-id"
 PARENT_SPAN_ID_HEADER = "context-parent-span-id"
+AGENT_PATH_HEADER = "context-agent-path"
 
 class ContextInterceptor(Interceptor):
     """Main interceptor that enables context threading through Temporal."""
@@ -90,6 +92,7 @@ def start_activity(self, input: StartActivityInput) -> workflow.ActivityHandle:
             task_id = getattr(workflow_instance, '_task_id', None)
             trace_id = getattr(workflow_instance, '_trace_id', None)
             parent_span_id = getattr(workflow_instance, '_parent_span_id', None)
+            agent_path = getattr(workflow_instance, '_agent_path', None)
 
             if task_id and trace_id and parent_span_id:
                 if not input.headers:
@@ -101,6 +104,14 @@ def start_activity(self, input: StartActivityInput) -> workflow.ActivityHandle:
                 logger.debug(f"[OutboundInterceptor] Added task_id, trace_id, and parent_span_id to activity headers: {task_id}, {trace_id}, {parent_span_id}")
             else:
                 logger.warning("[OutboundInterceptor] No _task_id, _trace_id, or _parent_span_id found in workflow instance")
+
+            # Independent of the three-tuple gate: a child-workflow sub-agent sets
+            # only _agent_path (task_id/trace/span come from the parent), and events
+            # from any agent need it to be attributable in a shared task stream.
+            if agent_path is not None:
+                if not input.headers:
+                    input.headers = {}
+                input.headers[AGENT_PATH_HEADER] = self._payload_converter.to_payload(agent_path)  # type: ignore[index]
         except Exception as e:
             logger.error(f"[OutboundInterceptor] Failed to get task_id, trace_id, or parent_span_id from workflow instance: {e}")
 
@@ -139,6 +150,15 @@ async def execute_activity(self, input: ExecuteActivityInput) -> Any:
         else:
             logger.debug("[ActivityInterceptor] No task_id, trace_id, or parent_span_id in headers")
 
+        # Threaded independently — present only when the emitting agent set it.
+        if input.headers and AGENT_PATH_HEADER in input.headers:
+            # Any (not str) — the value is str or list[str]; Any passes the decoded
+            # JSON through untouched instead of coercing to one variant.
+            agent_path_value = self._payload_converter.from_payload(
+                input.headers[AGENT_PATH_HEADER], Any
+            )
+            streaming_agent_path.set(agent_path_value)
+
         try:
             # Execute the activity
             # The TemporalStreamingModel can now read streaming_task_id.get()
@@ -149,5 +169,6 @@ async def execute_activity(self, input: ExecuteActivityInput) -> Any:
             streaming_task_id.set(None)
             streaming_trace_id.set(None)
             streaming_parent_span_id.set(None)
+            streaming_agent_path.set(None)
             logger.debug("[ActivityInterceptor] Cleared task_id, trace_id, and parent_span_id from context")
 

src/agentex/lib/core/temporal/plugins/openai_agents/models/temporal_streaming_model.py

@@ -71,6 +71,7 @@
 from agentex.lib.core.temporal.plugins.openai_agents.interceptors.context_interceptor import (
     streaming_task_id,
     streaming_trace_id,
+    streaming_agent_path,
     streaming_parent_span_id,
 )
 
@@ -635,6 +636,7 @@ async def get_response(
         task_id = streaming_task_id.get()
         trace_id = streaming_trace_id.get()
         parent_span_id = streaming_parent_span_id.get()
+        agent_path = streaming_agent_path.get()
 
         if not task_id or not trace_id or not parent_span_id:
             raise ValueError("task_id, trace_id, and parent_span_id are required for streaming with Responses API")
@@ -843,6 +845,7 @@ async def get_response(
                                         style="active",
                                     ),
                                     streaming_mode=self.streaming_mode,
+                                    agent_path=agent_path,
                                 ).__aenter__()
                         elif item and getattr(item, 'type', None) == 'function_call':
                             # Open a streaming context per function call so argument
@@ -859,6 +862,7 @@ async def get_response(
                                     arguments={},
                                 ),
                                 streaming_mode=self.streaming_mode,
+                                agent_path=agent_path,
                             ).__aenter__()
                             function_calls_in_progress[output_index] = {
                                 'id': getattr(item, 'id', ''),
@@ -879,6 +883,7 @@ async def get_response(
                                     format="markdown",
                                 ),
                                 streaming_mode=self.streaming_mode,
+                                agent_path=agent_path,
                             ).__aenter__()
 
                     elif isinstance(event, ResponseFunctionCallArgumentsDeltaEvent):

src/agentex/types/task_message.py

@@ -1,6 +1,6 @@
 # File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
 
-from typing import Optional
+from typing import List, Union, Optional
 from datetime import datetime
 from typing_extensions import Literal
 
@@ -27,6 +27,15 @@ class TaskMessage(BaseModel):
     task_id: str
     """ID of the task this message belongs to"""
 
+    # MANUAL PATCH — mirror into the upstream OpenAPI spec (see .stats.yml
+    # openapi_spec_url) or the next Stainless regen drops it.
+    agent_path: Optional[Union[str, List[str]]] = None
+    """Identifier of the agent that emitted this message.
+
+    A single agent id, or a root->emitter path (e.g. ["researcher", "subagent-abc"])
+    when nested sub-agents share one task stream. Consumers group/filter events by it.
+    """
+
     id: Optional[str] = None
     """The task message's unique id"""