diff --git a/.gitignore b/.gitignore
index 4994e9e2fe..5fa222ec40 100644
--- a/.gitignore
+++ b/.gitignore
@@ -235,3 +235,7 @@ python/dotnet-ref
 
 # Generated filtered solution files (created by eng/scripts/New-FilteredSolution.ps1)
 dotnet/filtered-*.slnx
+
+# Local tool state
+.omc/
+.omx/
diff --git a/python/packages/core/agent_framework/_workflows/_agent.py b/python/packages/core/agent_framework/_workflows/_agent.py
index 2fd3f35213..84c6ffafef 100644
--- a/python/packages/core/agent_framework/_workflows/_agent.py
+++ b/python/packages/core/agent_framework/_workflows/_agent.py
@@ -300,7 +300,9 @@ async def _run_impl(
             function_invocation_kwargs=function_invocation_kwargs,
             client_kwargs=client_kwargs,
         ):
-            if event.type == "output" or event.type == "request_info":
+            if event.type in ("output", "request_info") or (
+                event.type == "data" and isinstance(event.data, (AgentResponse, AgentResponseUpdate))
+            ):
                 output_events.append(event)
 
         result = self._convert_workflow_events_to_agent_response(response_id, output_events)
@@ -527,7 +529,14 @@ def _convert_workflow_events_to_agent_response(
                 )
                 raw_representations.append(output_event)
             else:
+                # `data` events carry intermediate participant responses (e.g., orchestration
+                # agents emitting via emit_data_events). Reframe their text content as
+                # `text_reasoning` so consumers can render them like agent thinking, mirroring
+                # how reasoning-capable agents (Claude thinking, OpenAI reasoning) already
+                # surface intermediate content. `output` events pass through unchanged.
+                as_reasoning = output_event.type == "data"
                 data = output_event.data
+
                 if isinstance(data, AgentResponseUpdate):
                     # We cannot support AgentResponseUpdate in non-streaming mode. This is because the message
                     # sequence cannot be guaranteed when there are streaming updates in between non-streaming
@@ -538,7 +547,7 @@ def _convert_workflow_events_to_agent_response(
                     )
 
                 if isinstance(data, AgentResponse):
-                    messages.extend(data.messages)
+                    messages.extend(self._msg_maybe_reasoning(m, as_reasoning=as_reasoning) for m in data.messages)
                     raw_representations.append(data.raw_representation)
                     merged_usage = add_usage_details(merged_usage, data.usage_details)
                     latest_created_at = (
@@ -549,16 +558,18 @@ def _convert_workflow_events_to_agent_response(
                         else latest_created_at
                     )
                 elif isinstance(data, Message):
-                    messages.append(data)
+                    messages.append(self._msg_maybe_reasoning(data, as_reasoning=as_reasoning))
                     raw_representations.append(data.raw_representation)
                 elif is_instance_of(data, list[Message]):
                     chat_messages = cast(list[Message], data)
-                    messages.extend(chat_messages)
+                    messages.extend(self._msg_maybe_reasoning(m, as_reasoning=as_reasoning) for m in chat_messages)
                     raw_representations.append(data)
                 else:
                     contents = self._extract_contents(data)
                     if not contents:
                         continue
+                    if as_reasoning:
+                        contents = self._rewrite_text_to_reasoning(contents)
 
                     messages.append(
                         Message(
@@ -618,33 +629,47 @@ def _convert_workflow_event_to_agent_response_updates(
     ) -> list[AgentResponseUpdate]:
         """Convert a workflow event to a list of AgentResponseUpdate objects.
 
-        Events with type='output' and type='request_info' are processed.
-        Other workflow events are ignored as they are workflow-internal.
-
-        For 'output' events, AgentExecutor yields AgentResponseUpdate for streaming updates
-        via ctx.yield_output(). This method converts those to agent response updates.
+        Processes `output` and `request_info` events, plus `data` events carrying
+        `AgentResponse` or `AgentResponseUpdate` (emitted by orchestrations to surface
+        intermediate participants when `intermediate_outputs=True`). Other event types
+        are workflow-internal and ignored.
 
         Returns:
             A list of AgentResponseUpdate objects. Empty list if the event is not relevant.
         """
-        if event.type == "output":
-            # Convert workflow output to agent response updates.
-            # Handle different data types appropriately.
-            data = event.data
+        data: Any = event.data
+        if event.type == "output" or (event.type == "data" and isinstance(data, (AgentResponse, AgentResponseUpdate))):
+            # `data` events carry intermediate participant content (e.g., orchestration agents
+            # via emit_data_events). Reframe their text content as `text_reasoning` so consumers
+            # render them as agent thinking. `output` events pass through unchanged.
+            as_reasoning = event.type == "data"
             executor_id = event.executor_id
 
+            def _contents(src: Sequence[Content]) -> list[Content]:
+                return self._rewrite_text_to_reasoning(src) if as_reasoning else list(src)
+
             if isinstance(data, AgentResponseUpdate):
-                # Pass through AgentResponseUpdate directly (streaming from AgentExecutor)
-                if not data.author_name:
-                    data.author_name = executor_id
-                return [data]
+                # Construct a fresh AgentResponseUpdate so we don't mutate a payload
+                # that AgentExecutor (and the data-event publisher) still hold references
+                # to in their `updates` list / output channel.
+                return [
+                    AgentResponseUpdate(
+                        contents=_contents(data.contents),
+                        role=data.role,
+                        author_name=data.author_name or executor_id,
+                        response_id=data.response_id,
+                        message_id=data.message_id,
+                        created_at=data.created_at,
+                        raw_representation=data.raw_representation,
+                    )
+                ]
             if isinstance(data, AgentResponse):
                 # Convert each message in AgentResponse to an AgentResponseUpdate
                 updates: list[AgentResponseUpdate] = []
                 for msg in data.messages:
                     updates.append(
                         AgentResponseUpdate(
-                            contents=list(msg.contents),
+                            contents=_contents(msg.contents),
                             role=msg.role,
                             author_name=msg.author_name or executor_id,
                             response_id=data.response_id or response_id,
@@ -658,7 +683,7 @@ def _convert_workflow_event_to_agent_response_updates(
             if isinstance(data, Message):
                 return [
                     AgentResponseUpdate(
-                        contents=list(data.contents),
+                        contents=_contents(data.contents),
                         role=data.role,
                         author_name=data.author_name or executor_id,
                         response_id=response_id,
@@ -674,7 +699,7 @@ def _convert_workflow_event_to_agent_response_updates(
                 for msg in chat_messages:
                     updates.append(
                         AgentResponseUpdate(
-                            contents=list(msg.contents),
+                            contents=_contents(msg.contents),
                             role=msg.role,
                             author_name=msg.author_name or executor_id,
                             response_id=response_id,
@@ -687,6 +712,8 @@ def _convert_workflow_event_to_agent_response_updates(
             contents = self._extract_contents(data)
             if not contents:
                 return []
+            if as_reasoning:
+                contents = self._rewrite_text_to_reasoning(contents)
             return [
                 AgentResponseUpdate(
                     contents=contents,
@@ -791,6 +818,52 @@ def _extract_contents(self, data: Any) -> list[Content]:
             return [Content.from_text(text=data)]
         return [Content.from_text(text=str(data))]
 
+    @staticmethod
+    def _rewrite_text_to_reasoning(contents: Sequence[Content]) -> list[Content]:
+        """Rewrite TextContent blocks as TextReasoningContent.
+
+        Used by WorkflowAgent to reframe content arriving on the workflow's `data` channel —
+        e.g., intermediate participants in an orchestration — as reasoning content from the
+        perspective of the wrapped workflow agent. This aligns workflow-as-agent intermediate
+        output with how reasoning-capable agents (Claude thinking, OpenAI reasoning) already
+        emit thinking content, so consumers can use one rendering path.
+
+        Non-text content (function calls, results, already-reasoning text, hosted files, etc.)
+        passes through unchanged.
+        """
+        rewritten: list[Content] = []
+        for content in contents:
+            if content.type == "text":
+                rewritten.append(
+                    Content.from_text_reasoning(
+                        id=content.id,
+                        text=content.text,
+                        annotations=content.annotations,
+                        additional_properties=content.additional_properties,
+                        raw_representation=content.raw_representation,
+                    )
+                )
+            else:
+                rewritten.append(content)
+        return rewritten
+
+    @classmethod
+    def _msg_as_reasoning(cls, msg: Message) -> Message:
+        """Return a copy of `msg` with text content rewritten as reasoning content."""
+        return Message(
+            role=msg.role,
+            contents=cls._rewrite_text_to_reasoning(msg.contents),
+            author_name=msg.author_name,
+            message_id=msg.message_id,
+            additional_properties=msg.additional_properties,
+            raw_representation=msg.raw_representation,
+        )
+
+    @classmethod
+    def _msg_maybe_reasoning(cls, msg: Message, *, as_reasoning: bool) -> Message:
+        """Conditional `_msg_as_reasoning`: rewrite when `as_reasoning` is True, pass through otherwise."""
+        return cls._msg_as_reasoning(msg) if as_reasoning else msg
+
     class _ResponseState(TypedDict):
         """State for grouping response updates by message_id."""
 
diff --git a/python/packages/core/agent_framework/_workflows/_agent_executor.py b/python/packages/core/agent_framework/_workflows/_agent_executor.py
index 626a02199b..91c379fed6 100644
--- a/python/packages/core/agent_framework/_workflows/_agent_executor.py
+++ b/python/packages/core/agent_framework/_workflows/_agent_executor.py
@@ -15,6 +15,7 @@
 from .._types import AgentResponse, AgentResponseUpdate, Message, ResponseStream
 from ._agent_utils import resolve_agent_id
 from ._const import GLOBAL_KWARGS_KEY, WORKFLOW_RUN_KWARGS_KEY
+from ._events import WorkflowEvent
 from ._executor import Executor, handler
 from ._message_utils import normalize_messages_input
 from ._request_info_mixin import response_handler
@@ -141,6 +142,7 @@ def __init__(
         id: str | None = None,
         context_mode: Literal["full", "last_agent", "custom"] | None = None,
         context_filter: Callable[[list[Message]], list[Message]] | None = None,
+        emit_data_events: bool = False,
     ):
         """Initialize the executor with a unique identifier.
 
@@ -158,6 +160,10 @@ def __init__(
                    as context for the agent run.
             context_filter: An optional function for filtering conversation context when context_mode is set
                 to "custom".
+            emit_data_events: When True, additionally emits `data` events (via
+                `WorkflowEvent.emit`) carrying each AgentResponse / AgentResponseUpdate alongside
+                the existing `output` events. Orchestrations use this to surface intermediate
+                participants while reserving `output` events for the workflow's final answer.
         """
         # Prefer provided id; else use agent.name if present; else generate deterministic prefix
         exec_id = id or resolve_agent_id(agent)
@@ -183,6 +189,8 @@ def __init__(
         if self._context_mode == "custom" and not self._context_filter:
             raise ValueError("context_filter must be provided when context_mode is set to 'custom'.")
 
+        self._emit_data_events = emit_data_events
+
     @property
     def agent(self) -> SupportsAgentRun:
         """Get the underlying agent wrapped by this executor."""
@@ -429,6 +437,8 @@ async def _run_agent(self, ctx: WorkflowContext[Never, AgentResponse]) -> AgentR
             client_kwargs=client_kwargs,
         )
         await ctx.yield_output(response)
+        if self._emit_data_events:
+            await ctx.add_event(WorkflowEvent.emit(self.id, response))
 
         # Handle any user input requests
         if response.user_input_requests:
@@ -472,6 +482,8 @@ async def _run_agent_streaming(self, ctx: WorkflowContext[Never, AgentResponseUp
         async for update in stream:
             updates.append(update)
             await ctx.yield_output(update)
+            if self._emit_data_events:
+                await ctx.add_event(WorkflowEvent.emit(self.id, update))
             if update.user_input_requests:
                 streamed_user_input_requests.extend(update.user_input_requests)
 
diff --git a/python/packages/core/agent_framework/_workflows/_workflow_executor.py b/python/packages/core/agent_framework/_workflows/_workflow_executor.py
index afb6145251..43d841ac18 100644
--- a/python/packages/core/agent_framework/_workflows/_workflow_executor.py
+++ b/python/packages/core/agent_framework/_workflows/_workflow_executor.py
@@ -566,6 +566,13 @@ async def _process_workflow_result(
         else:
             await asyncio.gather(*[ctx.send_message(output) for output in outputs])
 
+        # Forward data events from the sub-workflow so that intermediate
+        # observations (e.g. emit_data_events from AgentExecutor) are
+        # visible in the parent workflow's event stream.
+        data_events = [event for event in result if isinstance(event, WorkflowEvent) and event.type == "data"]
+        for data_event in data_events:
+            await ctx.add_event(WorkflowEvent.emit(data_event.executor_id or "", data_event.data))
+
         # Process request info events
         for event in request_info_events:
             request_id = event.request_id
diff --git a/python/packages/core/tests/workflow/test_agent_executor.py b/python/packages/core/tests/workflow/test_agent_executor.py
index 5ffd60aa55..1235ff3f2c 100644
--- a/python/packages/core/tests/workflow/test_agent_executor.py
+++ b/python/packages/core/tests/workflow/test_agent_executor.py
@@ -699,3 +699,61 @@ async def test_resolve_executor_kwargs_empty_per_executor_does_not_fallback_to_g
     resolved = {"exec_a": {}, GLOBAL_KWARGS_KEY: {"global_key": "global_val"}}
     result = executor._resolve_executor_kwargs(resolved)  # pyright: ignore[reportPrivateUsage]
     assert result == {}
+
+
+async def test_emit_data_events_mirrors_yield_output_non_streaming() -> None:
+    """When emit_data_events=True, AgentExecutor emits a data event with the AgentResponse."""
+    agent = _CountingAgent(id="agent_a", name="AgentA")
+    executor = AgentExecutor(agent, id="exec_a", emit_data_events=True)
+    workflow = WorkflowBuilder(start_executor=executor).build()
+
+    output_events: list[WorkflowEvent[Any]] = []
+    data_events: list[WorkflowEvent[Any]] = []
+    for event in await workflow.run("hello"):
+        if event.type == "output":
+            output_events.append(event)
+        elif event.type == "data":
+            data_events.append(event)
+
+    # Output event still emitted (existing behavior unchanged)
+    assert len(output_events) == 1
+    assert isinstance(output_events[0].data, AgentResponse)
+    # Plus a parallel data event with the same AgentResponse payload
+    assert len(data_events) == 1
+    assert data_events[0].executor_id == "exec_a"
+    assert isinstance(data_events[0].data, AgentResponse)
+    assert data_events[0].data.messages[0].text == output_events[0].data.messages[0].text
+
+
+async def test_emit_data_events_mirrors_yield_output_streaming() -> None:
+    """When emit_data_events=True and streaming, data events accompany each AgentResponseUpdate."""
+    agent = _CountingAgent(id="agent_a", name="AgentA")
+    executor = AgentExecutor(agent, id="exec_a", emit_data_events=True)
+    workflow = WorkflowBuilder(start_executor=executor).build()
+
+    output_updates: list[WorkflowEvent[Any]] = []
+    data_updates: list[WorkflowEvent[Any]] = []
+    async for event in workflow.run("hello", stream=True):
+        if event.type == "output":
+            output_updates.append(event)
+        elif event.type == "data":
+            data_updates.append(event)
+
+    assert output_updates and all(isinstance(e.data, AgentResponseUpdate) for e in output_updates)
+    assert len(data_updates) == len(output_updates)
+    assert all(isinstance(e.data, AgentResponseUpdate) for e in data_updates)
+    assert all(e.executor_id == "exec_a" for e in data_updates)
+
+
+async def test_emit_data_events_default_false_no_data_events() -> None:
+    """When emit_data_events is not set, no extra data events are emitted (default behavior)."""
+    agent = _CountingAgent(id="agent_a", name="AgentA")
+    executor = AgentExecutor(agent, id="exec_a")  # default: emit_data_events=False
+    workflow = WorkflowBuilder(start_executor=executor).build()
+
+    data_events: list[WorkflowEvent[Any]] = []
+    for event in await workflow.run("hello"):
+        if event.type == "data":
+            data_events.append(event)
+
+    assert data_events == []
diff --git a/python/packages/core/tests/workflow/test_workflow_agent.py b/python/packages/core/tests/workflow/test_workflow_agent.py
index 0101a6e8a5..666a88e2c7 100644
--- a/python/packages/core/tests/workflow/test_workflow_agent.py
+++ b/python/packages/core/tests/workflow/test_workflow_agent.py
@@ -23,6 +23,7 @@
     WorkflowAgent,
     WorkflowBuilder,
     WorkflowContext,
+    WorkflowEvent,
     executor,
     handler,
     response_handler,
@@ -1562,3 +1563,216 @@ def test_merge_updates_function_result_no_matching_call(self):
 
         # Order: text (user), text (assistant), function_result (orphan at end)
         assert content_types == ["text", "text", "function_result"]
+
+
+class _ReasoningEmittingExecutor(Executor):
+    """Test executor that emits a `data` event followed by an `output` event.
+
+    Mirrors the pattern AgentExecutor(emit_data_events=True) uses: a data event surfaces
+    intermediate observation, an output event carries the workflow's terminal answer.
+    Used to validate WorkflowAgent's data → text_reasoning conversion in isolation.
+    """
+
+    def __init__(
+        self,
+        id: str,
+        intermediate_data: AgentResponse | Message | list[Message],
+        terminal_output: AgentResponse | Message | list[Message],
+    ):
+        super().__init__(id=id)
+        self._intermediate_data = intermediate_data
+        self._terminal_output = terminal_output
+
+    @handler
+    async def handle(
+        self,
+        message: list[Message],
+        ctx: WorkflowContext[Any, Any],
+    ) -> None:
+        await ctx.add_event(WorkflowEvent.emit(self.id, self._intermediate_data))
+        await ctx.yield_output(self._terminal_output)
+
+
+class TestWorkflowAgentReasoningHelpers:
+    """Tests for WorkflowAgent._rewrite_text_to_reasoning and _msg_as_reasoning helpers."""
+
+    def test_rewrite_text_to_reasoning_converts_text(self) -> None:
+        """Text content blocks are converted to text_reasoning, preserving id and text."""
+        text = Content.from_text(text="hello world", additional_properties={"src": "agent1"})
+        result = WorkflowAgent._rewrite_text_to_reasoning([text])
+        assert len(result) == 1
+        assert result[0].type == "text_reasoning"
+        assert result[0].text == "hello world"  # type: ignore[attr-defined]
+        assert result[0].additional_properties.get("src") == "agent1"
+
+    def test_rewrite_text_to_reasoning_passes_through_function_call(self) -> None:
+        """Non-text content (function calls, results) passes through unchanged."""
+        fc = Content.from_function_call(call_id="c1", name="my_tool", arguments={"x": 1})
+        result = WorkflowAgent._rewrite_text_to_reasoning([fc])
+        assert len(result) == 1
+        assert result[0] is fc  # same instance — passed through
+
+    def test_rewrite_text_to_reasoning_no_double_wrap(self) -> None:
+        """Already-reasoning content stays as text_reasoning (not wrapped again)."""
+        already = Content.from_text_reasoning(text="thinking")
+        result = WorkflowAgent._rewrite_text_to_reasoning([already])
+        assert len(result) == 1
+        assert result[0] is already  # same instance — only type=='text' is rewritten
+
+    def test_rewrite_text_to_reasoning_handles_mixed_content(self) -> None:
+        """Mixed content: only text blocks are rewritten; others pass through."""
+        text = Content.from_text(text="answer")
+        fc = Content.from_function_call(call_id="c1", name="my_tool", arguments={})
+        already = Content.from_text_reasoning(text="prior thinking")
+        result = WorkflowAgent._rewrite_text_to_reasoning([text, fc, already])
+        assert [c.type for c in result] == ["text_reasoning", "function_call", "text_reasoning"]
+        assert result[1] is fc
+        assert result[2] is already
+
+    def test_msg_as_reasoning_preserves_role_and_metadata(self) -> None:
+        """_msg_as_reasoning copies the message with rewritten contents but preserves all other fields."""
+        original = Message(
+            "assistant",
+            [Content.from_text(text="hi"), Content.from_function_call(call_id="c1", name="t", arguments={})],
+            author_name="agent1",
+            message_id="msg-123",
+            additional_properties={"meta": "value"},
+        )
+        new_msg = WorkflowAgent._msg_as_reasoning(original)
+        assert new_msg is not original
+        assert new_msg.role == "assistant"
+        assert new_msg.author_name == "agent1"
+        assert new_msg.message_id == "msg-123"
+        assert new_msg.additional_properties.get("meta") == "value"
+        assert [c.type for c in new_msg.contents] == ["text_reasoning", "function_call"]
+        # Original message is unmodified
+        assert [c.type for c in original.contents] == ["text", "function_call"]
+
+
+class TestWorkflowAgentDataEventReasoningConversion:
+    """End-to-end tests for as_agent() rewriting data event content as reasoning."""
+
+    async def test_data_event_text_becomes_reasoning_non_streaming(self) -> None:
+        """A data event carrying AgentResponse with text content surfaces as text_reasoning in as_agent()."""
+        intermediate = AgentResponse(messages=[Message("assistant", [Content.from_text(text="thinking step")])])
+        terminal = AgentResponse(messages=[Message("assistant", [Content.from_text(text="final answer")])])
+        exec_ = _ReasoningEmittingExecutor(id="exec_a", intermediate_data=intermediate, terminal_output=terminal)
+        agent = WorkflowBuilder(start_executor=exec_).build().as_agent()
+
+        response = await agent.run("go")
+
+        assert isinstance(response, AgentResponse)
+        all_types = [c.type for m in response.messages for c in m.contents]
+        # Intermediate content rewritten to text_reasoning; terminal stays text.
+        assert "text_reasoning" in all_types
+        assert "text" in all_types
+        reasoning_text = " ".join(
+            c.text or "" for m in response.messages for c in m.contents if c.type == "text_reasoning"
+        )
+        answer_text = " ".join(c.text or "" for m in response.messages for c in m.contents if c.type == "text")
+        assert reasoning_text == "thinking step"
+        assert answer_text == "final answer"
+
+    async def test_output_event_text_passes_through_non_streaming(self) -> None:
+        """An output event with text content passes through unchanged (not rewritten)."""
+
+        class _OutputOnly(Executor):
+            @handler
+            async def handle(self, message: list[Message], ctx: WorkflowContext[Any, Any]) -> None:
+                response = AgentResponse(messages=[Message("assistant", [Content.from_text(text="answer")])])
+                await ctx.yield_output(response)
+
+        agent = WorkflowBuilder(start_executor=_OutputOnly(id="solo")).build().as_agent()
+        response = await agent.run("go")
+        all_types = [c.type for m in response.messages for c in m.contents]
+        assert all_types == ["text"], "Output events must not be rewritten as reasoning"
+
+    async def test_data_event_with_mixed_content_only_text_rewritten(self) -> None:
+        """In a data event, only text content is rewritten; function_call/result pass through."""
+        intermediate = AgentResponse(
+            messages=[
+                Message(
+                    "assistant",
+                    [
+                        Content.from_text(text="reasoning"),
+                        Content.from_function_call(call_id="c1", name="search", arguments={"q": "x"}),
+                    ],
+                )
+            ]
+        )
+        terminal = AgentResponse(messages=[Message("assistant", [Content.from_text(text="done")])])
+        exec_ = _ReasoningEmittingExecutor(id="mix", intermediate_data=intermediate, terminal_output=terminal)
+        agent = WorkflowBuilder(start_executor=exec_).build().as_agent()
+        response = await agent.run("go")
+
+        # Find the message that has the function_call (the intermediate one)
+        intermediate_msg = next(m for m in response.messages if any(c.type == "function_call" for c in m.contents))
+        types = [c.type for c in intermediate_msg.contents]
+        assert types == ["text_reasoning", "function_call"]
+
+    async def test_data_event_already_reasoning_not_double_wrapped(self) -> None:
+        """A data event whose content is already text_reasoning surfaces unchanged (no double wrap)."""
+        intermediate = AgentResponse(
+            messages=[Message("assistant", [Content.from_text_reasoning(text="already thinking")])]
+        )
+        terminal = AgentResponse(messages=[Message("assistant", [Content.from_text(text="done")])])
+        exec_ = _ReasoningEmittingExecutor(id="reasoning_in", intermediate_data=intermediate, terminal_output=terminal)
+        agent = WorkflowBuilder(start_executor=exec_).build().as_agent()
+        response = await agent.run("go")
+
+        reasoning_blocks = [c for m in response.messages for c in m.contents if c.type == "text_reasoning"]
+        assert len(reasoning_blocks) == 1
+        assert reasoning_blocks[0].text == "already thinking"  # type: ignore[attr-defined]
+
+    async def test_data_event_text_becomes_reasoning_streaming(self) -> None:
+        """In streaming mode, AgentResponseUpdate from data events carries text_reasoning content."""
+        intermediate = AgentResponse(messages=[Message("assistant", [Content.from_text(text="midway")])])
+        terminal = AgentResponse(messages=[Message("assistant", [Content.from_text(text="end")])])
+        exec_ = _ReasoningEmittingExecutor(id="stream_x", intermediate_data=intermediate, terminal_output=terminal)
+        agent = WorkflowBuilder(start_executor=exec_).build().as_agent()
+
+        updates: list[AgentResponseUpdate] = []
+        async for update in agent.run("go", stream=True):
+            updates.append(update)
+
+        all_types = [c.type for u in updates for c in u.contents]
+        assert "text_reasoning" in all_types
+        assert "text" in all_types
+        # Reasoning chunk's text matches the intermediate
+        reasoning_chunks = [c for u in updates for c in u.contents if c.type == "text_reasoning"]
+        assert any((c.text or "") == "midway" for c in reasoning_chunks)
+        # Terminal text chunk matches
+        text_chunks = [c for u in updates for c in u.contents if c.type == "text"]
+        assert any((c.text or "") == "end" for c in text_chunks)
+
+    async def test_data_event_streaming_does_not_mutate_source_update(self) -> None:
+        """Reasoning rewriting must not mutate the AgentResponseUpdate the source emitted.
+
+        AgentExecutor (and other emit_data_events publishers) hold references to the
+        update in their local `updates` list and yielded output channel. Mutating
+        `data.contents` in place would silently corrupt the AgentResponse the executor
+        finalizes from those updates.
+        """
+        original_update = AgentResponseUpdate(
+            contents=[Content.from_text(text="reason")],
+            role="assistant",
+            author_name="agent_a",
+        )
+
+        class _SharedUpdateExecutor(Executor):
+            @handler
+            async def handle(self, message: list[Message], ctx: WorkflowContext[Any, Any]) -> None:
+                # Mirror what AgentExecutor(emit_data_events=True) does: emit the same
+                # update via both data and output channels.
+                await ctx.add_event(WorkflowEvent.emit(self.id, original_update))
+                await ctx.yield_output(
+                    AgentResponse(messages=[Message("assistant", [Content.from_text(text="final")])])
+                )
+
+        agent = WorkflowBuilder(start_executor=_SharedUpdateExecutor(id="src")).build().as_agent()
+        async for _ in agent.run("go", stream=True):
+            pass
+
+        # Source update content must be unchanged (still `text`, never rewritten to `text_reasoning`).
+        assert [c.type for c in original_update.contents] == ["text"]
+        assert original_update.author_name == "agent_a"  # not stamped with executor id
diff --git a/python/packages/orchestrations/agent_framework_orchestrations/_base_group_chat_orchestrator.py b/python/packages/orchestrations/agent_framework_orchestrations/_base_group_chat_orchestrator.py
index 86c85cc079..2cca41d9ee 100644
--- a/python/packages/orchestrations/agent_framework_orchestrations/_base_group_chat_orchestrator.py
+++ b/python/packages/orchestrations/agent_framework_orchestrations/_base_group_chat_orchestrator.py
@@ -12,7 +12,7 @@
 from dataclasses import dataclass
 from typing import Any, ClassVar, TypeAlias
 
-from agent_framework._types import Message
+from agent_framework._types import AgentResponse, Message
 from agent_framework._workflows._agent_executor import AgentExecutor, AgentExecutorRequest, AgentExecutorResponse
 from agent_framework._workflows._events import WorkflowEvent
 from agent_framework._workflows._executor import Executor, handler
@@ -351,8 +351,8 @@ async def _check_termination(self) -> bool:
             result = await result
         return result
 
-    async def _check_terminate_and_yield(self, ctx: WorkflowContext[Never, list[Message]]) -> bool:
-        """Check termination conditions and yield completion if met.
+    async def _check_terminate_and_yield(self, ctx: WorkflowContext[Never, AgentResponse]) -> bool:
+        """Check termination conditions and yield the completion message if met.
 
         Args:
             ctx: Workflow context for yielding output
@@ -362,8 +362,9 @@ async def _check_terminate_and_yield(self, ctx: WorkflowContext[Never, list[Mess
         """
         terminate = await self._check_termination()
         if terminate:
-            self._append_messages([self._create_completion_message(self.TERMINATION_CONDITION_MET_MESSAGE)])
-            await ctx.yield_output(self._full_conversation)
+            completion_message = self._create_completion_message(self.TERMINATION_CONDITION_MET_MESSAGE)
+            self._append_messages([completion_message])
+            await ctx.yield_output(AgentResponse(messages=[completion_message]))
             return True
 
         return False
@@ -490,8 +491,8 @@ def _check_round_limit(self) -> bool:
 
         return False
 
-    async def _check_round_limit_and_yield(self, ctx: WorkflowContext[Never, list[Message]]) -> bool:
-        """Check round limit and yield completion if reached.
+    async def _check_round_limit_and_yield(self, ctx: WorkflowContext[Never, AgentResponse]) -> bool:
+        """Check round limit and yield the max-rounds completion message if reached.
 
         Args:
             ctx: Workflow context for yielding output
@@ -501,8 +502,9 @@ async def _check_round_limit_and_yield(self, ctx: WorkflowContext[Never, list[Me
         """
         reach_max_rounds = self._check_round_limit()
         if reach_max_rounds:
-            self._append_messages([self._create_completion_message(self.MAX_ROUNDS_MET_MESSAGE)])
-            await ctx.yield_output(self._full_conversation)
+            completion_message = self._create_completion_message(self.MAX_ROUNDS_MET_MESSAGE)
+            self._append_messages([completion_message])
+            await ctx.yield_output(AgentResponse(messages=[completion_message]))
             return True
 
         return False
diff --git a/python/packages/orchestrations/agent_framework_orchestrations/_concurrent.py b/python/packages/orchestrations/agent_framework_orchestrations/_concurrent.py
index d73b7e322b..81cf6661f0 100644
--- a/python/packages/orchestrations/agent_framework_orchestrations/_concurrent.py
+++ b/python/packages/orchestrations/agent_framework_orchestrations/_concurrent.py
@@ -6,7 +6,7 @@
 from collections.abc import Callable, Sequence
 from typing import Any
 
-from agent_framework import Message, SupportsAgentRun
+from agent_framework import AgentResponse, Message, SupportsAgentRun
 from agent_framework._workflows._agent_executor import AgentExecutor, AgentExecutorRequest, AgentExecutorResponse
 from agent_framework._workflows._agent_utils import resolve_agent_id
 from agent_framework._workflows._checkpoint import CheckpointStorage
@@ -71,18 +71,20 @@ async def from_messages(
 
 
 class _AggregateAgentConversations(Executor):
-    """Aggregates agent responses and completes with combined ChatMessages.
+    """Aggregates agent responses and completes with a single AgentResponse.
 
-    Emits a list[Message] shaped as:
-      [ single_user_prompt?, agent1_final_assistant, agent2_final_assistant, ... ]
+    Emits an `AgentResponse` whose `messages` are the final assistant message from each
+    participant (one message per agent), in deterministic participant order matching
+    the fan-in `sources` configuration. The user prompt is intentionally not included —
+    that is part of the input, not the answer.
 
-    - Extracts a single user prompt (first user message seen across results).
-    - For each result, selects the final assistant message (prefers agent_response.messages).
-    - Avoids duplicating the same user message per agent.
+    For each participant the final assistant message is sourced from
+    `r.agent_response.messages`, falling back to scanning `r.full_conversation` for
+    pathological executors that did not populate the response.
     """
 
     @handler
-    async def aggregate(self, results: list[AgentExecutorResponse], ctx: WorkflowContext[Never, list[Message]]) -> None:
+    async def aggregate(self, results: list[AgentExecutorResponse], ctx: WorkflowContext[Never, AgentResponse]) -> None:
         if not results:
             logger.error("Concurrent aggregator received empty results list")
             raise ValueError("Aggregation failed: no results provided")
@@ -91,12 +93,10 @@ def _is_role(msg: Any, role: str) -> bool:
             r = getattr(msg, "role", None)
             if r is None:
                 return False
-            # Normalize both r and role to lowercase strings for comparison
             r_str = str(r).lower() if isinstance(r, str) or hasattr(r, "__str__") else r
             role_str = str(role).lower()
             return r_str == role_str
 
-        prompt_message: Message | None = None
         assistant_replies: list[Message] = []
 
         for r in results:
@@ -107,10 +107,6 @@ def _is_role(msg: Any, role: str) -> bool:
                 f"{len(resp_messages)} response msgs, {len(r.full_conversation)} conversation msgs"
             )
 
-            # Capture a single user prompt (first encountered across any conversation)
-            if prompt_message is None:
-                prompt_message = next((m for m in r.full_conversation if _is_role(m, "user")), None)
-
             # Pick the final assistant message from the response; fallback to conversation search
             final_assistant = next((m for m in reversed(resp_messages) if _is_role(m, "assistant")), None)
             if final_assistant is None:
@@ -127,14 +123,7 @@ def _is_role(msg: Any, role: str) -> bool:
             logger.error(f"Aggregation failed: no assistant replies found across {len(results)} results")
             raise RuntimeError("Aggregation failed: no assistant replies found")
 
-        output: list[Message] = []
-        if prompt_message is not None:
-            output.append(prompt_message)
-        else:
-            logger.warning("No user prompt found in any conversation; emitting assistants only")
-        output.extend(assistant_replies)
-
-        await ctx.yield_output(output)
+        await ctx.yield_output(AgentResponse(messages=assistant_replies))
 
 
 class _CallbackAggregator(Executor):
@@ -190,7 +179,8 @@ class ConcurrentBuilder:
 
         from agent_framework_orchestrations import ConcurrentBuilder
 
-        # Minimal: use default aggregator (returns list[Message])
+        # Minimal: use default aggregator (yields one AgentResponse with one assistant
+        # message per participant)
         workflow = ConcurrentBuilder(participants=[agent1, agent2, agent3]).build()
 
 
@@ -351,7 +341,13 @@ def with_request_info(
         return self
 
     def _resolve_participants(self) -> list[Executor]:
-        """Resolve participant instances into Executor objects."""
+        """Resolve participant instances into Executor objects.
+
+        When `intermediate_outputs=True`, every wrapped agent is constructed with
+        `emit_data_events=True` so its individual response surfaces as a `data`
+        event without polluting the single `output` event reserved for the aggregator's
+        final answer.
+        """
         if not self._participants:
             raise ValueError("No participants provided. Pass participants to the constructor.")
 
@@ -366,9 +362,9 @@ def _resolve_participants(self) -> list[Executor]:
                     not self._request_info_filter or resolve_agent_id(p) in self._request_info_filter
                 ):
                     # Handle request info enabled agents
-                    executors.append(AgentApprovalExecutor(p))
+                    executors.append(AgentApprovalExecutor(p, emit_data_events=self._intermediate_outputs))
                 else:
-                    executors.append(AgentExecutor(p))
+                    executors.append(AgentExecutor(p, emit_data_events=self._intermediate_outputs))
             else:
                 raise TypeError(f"Participants must be SupportsAgentRun or Executor instances. Got {type(p).__name__}.")
 
@@ -383,7 +379,7 @@ def build(self) -> Workflow:
         - If request info is enabled, the orchestration emits a request info event with outputs from all participants
             before sending the outputs to the aggregator
         - Aggregator yields output and the workflow becomes idle. The output is either:
-          - list[Message] (default aggregator: one user + one assistant per agent)
+          - AgentResponse (default aggregator: one assistant message per participant)
           - custom payload from the provided aggregator
 
         Returns:
@@ -408,7 +404,7 @@ def build(self) -> Workflow:
         builder = WorkflowBuilder(
             start_executor=dispatcher,
             checkpoint_storage=self._checkpoint_storage,
-            output_executors=[aggregator] if not self._intermediate_outputs else None,
+            output_executors=[aggregator],
         )
         # Fan-out for parallel execution
         builder.add_fan_out_edges(dispatcher, participants)
diff --git a/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py b/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py
index 4f1c2f832a..cf71760afa 100644
--- a/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py
+++ b/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py
@@ -29,7 +29,7 @@
 from dataclasses import dataclass
 from typing import Any, ClassVar, cast
 
-from agent_framework import Agent, AgentSession, Message, SupportsAgentRun
+from agent_framework import Agent, AgentResponse, AgentSession, Message, SupportsAgentRun
 from agent_framework._workflows._agent_executor import AgentExecutor, AgentExecutorRequest, AgentExecutorResponse
 from agent_framework._workflows._agent_utils import resolve_agent_id
 from agent_framework._workflows._checkpoint import CheckpointStorage
@@ -169,7 +169,7 @@ async def _handle_messages(
         """Initialize orchestrator state and start the conversation loop."""
         self._append_messages(messages)
         # Termination condition will also be applied to the input messages
-        if await self._check_terminate_and_yield(cast(WorkflowContext[Never, list[Message]], ctx)):
+        if await self._check_terminate_and_yield(cast(WorkflowContext[Never, AgentResponse], ctx)):
             return
 
         next_speaker = await self._get_next_speaker()
@@ -198,9 +198,9 @@ async def _handle_response(
         messages = clean_conversation_for_handoff(messages)
         self._append_messages(messages)
 
-        if await self._check_terminate_and_yield(cast(WorkflowContext[Never, list[Message]], ctx)):
+        if await self._check_terminate_and_yield(cast(WorkflowContext[Never, AgentResponse], ctx)):
             return
-        if await self._check_round_limit_and_yield(cast(WorkflowContext[Never, list[Message]], ctx)):
+        if await self._check_round_limit_and_yield(cast(WorkflowContext[Never, AgentResponse], ctx)):
             return
 
         next_speaker = await self._get_next_speaker()
@@ -332,13 +332,13 @@ async def _handle_messages(
         """Initialize orchestrator state and start the conversation loop."""
         self._append_messages(messages)
         # Termination condition will also be applied to the input messages
-        if await self._check_terminate_and_yield(cast(WorkflowContext[Never, list[Message]], ctx)):
+        if await self._check_terminate_and_yield(cast(WorkflowContext[Never, AgentResponse], ctx)):
             return
 
         agent_orchestration_output = await self._invoke_agent()
         if await self._check_agent_terminate_and_yield(
             agent_orchestration_output,
-            cast(WorkflowContext[Never, list[Message]], ctx),
+            cast(WorkflowContext[Never, AgentResponse], ctx),
         ):
             return
 
@@ -366,15 +366,15 @@ async def _handle_response(
         # Remove tool-related content to prevent API errors from empty messages
         messages = clean_conversation_for_handoff(messages)
         self._append_messages(messages)
-        if await self._check_terminate_and_yield(cast(WorkflowContext[Never, list[Message]], ctx)):
+        if await self._check_terminate_and_yield(cast(WorkflowContext[Never, AgentResponse], ctx)):
             return
-        if await self._check_round_limit_and_yield(cast(WorkflowContext[Never, list[Message]], ctx)):
+        if await self._check_round_limit_and_yield(cast(WorkflowContext[Never, AgentResponse], ctx)):
             return
 
         agent_orchestration_output = await self._invoke_agent()
         if await self._check_agent_terminate_and_yield(
             agent_orchestration_output,
-            cast(WorkflowContext[Never, list[Message]], ctx),
+            cast(WorkflowContext[Never, AgentResponse], ctx),
         ):
             return
 
@@ -522,9 +522,9 @@ async def _invoke_agent_helper(conversation: list[Message]) -> AgentOrchestratio
     async def _check_agent_terminate_and_yield(
         self,
         agent_orchestration_output: AgentOrchestrationOutput,
-        ctx: WorkflowContext[Never, list[Message]],
+        ctx: WorkflowContext[Never, AgentResponse],
     ) -> bool:
-        """Check if the agent requested termination and yield completion if so.
+        """Yield the orchestrator's completion `AgentResponse` if termination was requested.
 
         Args:
             agent_orchestration_output: Output from the orchestrator agent
@@ -536,8 +536,9 @@ async def _check_agent_terminate_and_yield(
             final_message = (
                 agent_orchestration_output.final_message or "The conversation has been terminated by the agent."
             )
-            self._append_messages([self._create_completion_message(final_message)])
-            await ctx.yield_output(self._full_conversation)
+            completion_message = self._create_completion_message(final_message)
+            self._append_messages([completion_message])
+            await ctx.yield_output(AgentResponse(messages=[completion_message]))
             return True
 
         return False
@@ -963,9 +964,9 @@ def _resolve_participants(self) -> list[Executor]:
                     not self._request_info_filter or resolve_agent_id(participant) in self._request_info_filter
                 ):
                     # Handle request info enabled agents
-                    executors.append(AgentApprovalExecutor(participant))
+                    executors.append(AgentApprovalExecutor(participant, emit_data_events=self._intermediate_outputs))
                 else:
-                    executors.append(AgentExecutor(participant))
+                    executors.append(AgentExecutor(participant, emit_data_events=self._intermediate_outputs))
             else:
                 raise TypeError(
                     f"Participants must be SupportsAgentRun or Executor instances. Got {type(participant).__name__}."
@@ -991,7 +992,7 @@ def build(self) -> Workflow:
         workflow_builder = WorkflowBuilder(
             start_executor=orchestrator,
             checkpoint_storage=self._checkpoint_storage,
-            output_executors=[orchestrator] if not self._intermediate_outputs else None,
+            output_executors=[orchestrator],
         )
         for participant in participants:
             # Orchestrator and participant bi-directional edges
diff --git a/python/packages/orchestrations/agent_framework_orchestrations/_handoff.py b/python/packages/orchestrations/agent_framework_orchestrations/_handoff.py
index c3e156096c..f555ab89b0 100644
--- a/python/packages/orchestrations/agent_framework_orchestrations/_handoff.py
+++ b/python/packages/orchestrations/agent_framework_orchestrations/_handoff.py
@@ -352,7 +352,7 @@ async def _run_agent_and_emit(self, ctx: WorkflowContext[Any, Any]) -> None:
         self._full_conversation.extend(self._cache.copy())
 
         # Check termination condition before running the agent
-        if await self._check_terminate_and_yield(ctx):
+        if await self._should_terminate():
             return
 
         # Run the agent
@@ -410,7 +410,7 @@ async def _run_agent_and_emit(self, ctx: WorkflowContext[Any, Any]) -> None:
         # Re-evaluate termination after appending and broadcasting this response.
         # Without this check, workflows that become terminal due to the latest assistant
         # message would still emit request_info and require an unnecessary extra resume.
-        if await self._check_terminate_and_yield(ctx):
+        if await self._should_terminate():
             return
 
         # Handle case where no handoff was requested
@@ -447,10 +447,10 @@ async def handle_response(
             response: The user's response messages
             ctx: The workflow context
 
-        If the response is empty, it indicates termination of the handoff workflow.
+        If the response is empty, the handoff workflow terminates. Per-agent responses
+        already surfaced as `output` events; no terminal yield is needed.
         """
         if not response:
-            await ctx.yield_output(self._full_conversation)
             return
 
         # Broadcast the user response to all other agents
@@ -520,14 +520,12 @@ def _is_handoff_requested(self, response: AgentResponse) -> tuple[str, Message]
 
         return None
 
-    async def _check_terminate_and_yield(self, ctx: WorkflowContext[Any, Any]) -> bool:
-        """Check termination conditions and yield completion if met.
+    async def _should_terminate(self) -> bool:
+        """Pure predicate: return True iff the configured termination condition is satisfied.
 
-        Args:
-            ctx: Workflow context for yielding output
-
-        Returns:
-            True if termination condition met and output yielded, False otherwise
+        Per-agent responses already surface as `output` events as agents speak, so the
+        handoff workflow has no terminal yield to make — this method only decides whether
+        the workflow should stop iterating.
         """
         if self._termination_condition is None:
             return False
@@ -535,12 +533,7 @@ async def _check_terminate_and_yield(self, ctx: WorkflowContext[Any, Any]) -> bo
         terminated = self._termination_condition(self._full_conversation)
         if inspect.isawaitable(terminated):
             terminated = await terminated
-
-        if terminated:
-            await ctx.yield_output(self._full_conversation)
-            return True
-
-        return False
+        return bool(terminated)
 
     @override
     async def on_checkpoint_save(self) -> dict[str, Any]:
@@ -577,13 +570,15 @@ class HandoffBuilder:
     tool injection, and middleware — capabilities only available on ``Agent``.
 
     Outputs:
-    The final conversation history as a list of Message once the group chat completes.
+    Each agent's response surfaces as a workflow `output` event as it speaks; there is no
+    synthetic terminal event. Consumers iterating events see per-agent ``AgentResponse`` (or
+    ``AgentResponseUpdate`` while streaming) in conversation order. The workflow returns to
+    idle once the termination condition is met (or the user terminates an interactive run).
 
     Note:
     1. Agents in handoff workflows must be ``Agent`` instances and support local tool calls.
-    2. Handoff doesn't support intermediate outputs from agents. All outputs are returned as
-       they become available. This is because agents in handoff workflows are not considered
-       sub-agents of a central orchestrator, thus all outputs are directly emitted.
+    2. Because each agent's response is itself a workflow output, handoff has no separate
+       "intermediate outputs" channel — every per-agent response is the primary output.
     """
 
     def __init__(
diff --git a/python/packages/orchestrations/agent_framework_orchestrations/_magentic.py b/python/packages/orchestrations/agent_framework_orchestrations/_magentic.py
index 80031cd726..ada2354014 100644
--- a/python/packages/orchestrations/agent_framework_orchestrations/_magentic.py
+++ b/python/packages/orchestrations/agent_framework_orchestrations/_magentic.py
@@ -1057,7 +1057,7 @@ async def _run_inner_loop_helper(
         if self._magentic_context is None:
             raise RuntimeError("Context not initialized")
         # Check limits first
-        within_limits = await self._check_within_limits_or_complete(cast(WorkflowContext[Never, list[Message]], ctx))
+        within_limits = await self._check_within_limits_or_complete(cast(WorkflowContext[Never, AgentResponse], ctx))
         if not within_limits:
             return
 
@@ -1092,7 +1092,7 @@ async def _run_inner_loop_helper(
         # Check for task completion
         if self._progress_ledger.is_request_satisfied.answer:
             logger.info("Magentic Orchestrator: Task completed")
-            await self._prepare_final_answer(cast(WorkflowContext[Never, list[Message]], ctx))
+            await self._prepare_final_answer(cast(WorkflowContext[Never, AgentResponse], ctx))
             return
 
         # Check for stalling or looping
@@ -1116,7 +1116,7 @@ async def _run_inner_loop_helper(
 
         if next_speaker not in self._participant_registry.participants:
             logger.warning(f"Invalid next speaker: {next_speaker}")
-            await self._prepare_final_answer(cast(WorkflowContext[Never, list[Message]], ctx))
+            await self._prepare_final_answer(cast(WorkflowContext[Never, AgentResponse], ctx))
             return
 
         # Add instruction to conversation (assistant guidance)
@@ -1192,23 +1192,23 @@ async def _run_outer_loop(
         # Start inner loop
         await self._run_inner_loop(ctx)
 
-    async def _prepare_final_answer(self, ctx: WorkflowContext[Never, list[Message]]) -> None:
-        """Prepare the final answer using the manager."""
+    async def _prepare_final_answer(self, ctx: WorkflowContext[Never, AgentResponse]) -> None:
+        """Yield the manager's synthesized final answer as the workflow's `AgentResponse`."""
         if self._magentic_context is None:
             raise RuntimeError("Context not initialized")
 
         logger.info("Magentic Orchestrator: Preparing final answer")
         final_answer = await self._manager.prepare_final_answer(self._magentic_context.clone(deep=True))
 
-        # Emit a completed event for the workflow
-        await ctx.yield_output([final_answer])
+        await ctx.yield_output(AgentResponse(messages=[final_answer]))
 
         self._terminated = True
 
-    async def _check_within_limits_or_complete(self, ctx: WorkflowContext[Never, list[Message]]) -> bool:
+    async def _check_within_limits_or_complete(self, ctx: WorkflowContext[Never, AgentResponse]) -> bool:
         """Check if orchestrator is within operational limits.
 
-        If limits are exceeded, yield a termination message and mark the workflow as terminated.
+        If limits are exceeded, yield a termination AgentResponse and mark the workflow
+        as terminated.
 
         Args:
             ctx: The workflow context.
@@ -1229,15 +1229,12 @@ async def _check_within_limits_or_complete(self, ctx: WorkflowContext[Never, lis
             limit_type = "round" if hit_round_limit else "reset"
             logger.error(f"Magentic Orchestrator: Max {limit_type} count reached")
 
-            # Yield the full conversation with an indication of termination due to limits
-            await ctx.yield_output([
-                *self._magentic_context.chat_history,
-                Message(
-                    role="assistant",
-                    contents=[f"Workflow terminated due to reaching maximum {limit_type} count."],
-                    author_name=MAGENTIC_MANAGER_NAME,
-                ),
-            ])
+            termination_message = Message(
+                role="assistant",
+                contents=[f"Workflow terminated due to reaching maximum {limit_type} count."],
+                author_name=MAGENTIC_MANAGER_NAME,
+            )
+            await ctx.yield_output(AgentResponse(messages=[termination_message]))
             self._terminated = True
 
             return False
@@ -1316,7 +1313,7 @@ async def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
 class MagenticAgentExecutor(AgentExecutor):
     """Specialized AgentExecutor for Magentic agent participants."""
 
-    def __init__(self, agent: SupportsAgentRun) -> None:
+    def __init__(self, agent: SupportsAgentRun, *, emit_data_events: bool = False) -> None:
         """Initialize a Magentic Agent Executor.
 
         This executor wraps an SupportsAgentRun instance to be used as a participant
@@ -1324,13 +1321,14 @@ def __init__(self, agent: SupportsAgentRun) -> None:
 
         Args:
             agent: The agent instance to wrap.
+            emit_data_events: Forwarded to the base AgentExecutor.
 
         Notes: Magentic pattern requires a reset operation upon replanning. This executor
         extends the base AgentExecutor to handle resets appropriately. In order to handle
         resets, the agent threads and other states are reset when requested by the orchestrator.
         And because of this, MagenticAgentExecutor does not support custom threads.
         """
-        super().__init__(agent)
+        super().__init__(agent, emit_data_events=emit_data_events)
 
     @handler
     async def handle_magentic_reset(self, signal: MagenticResetSignal, ctx: WorkflowContext) -> None:
@@ -1741,7 +1739,7 @@ def _resolve_participants(self) -> list[Executor]:
             if isinstance(participant, Executor):
                 executors.append(participant)
             elif isinstance(participant, SupportsAgentRun):
-                executors.append(MagenticAgentExecutor(participant))
+                executors.append(MagenticAgentExecutor(participant, emit_data_events=self._intermediate_outputs))
             else:
                 raise TypeError(
                     f"Participants must be SupportsAgentRun or Executor instances. Got {type(participant).__name__}."
@@ -1760,7 +1758,7 @@ def build(self) -> Workflow:
         workflow_builder = WorkflowBuilder(
             start_executor=orchestrator,
             checkpoint_storage=self._checkpoint_storage,
-            output_executors=[orchestrator] if not self._intermediate_outputs else None,
+            output_executors=[orchestrator],
         )
         for participant in participants:
             # Orchestrator and participant bi-directional edges
diff --git a/python/packages/orchestrations/agent_framework_orchestrations/_orchestration_request_info.py b/python/packages/orchestrations/agent_framework_orchestrations/_orchestration_request_info.py
index e78d1bef14..128714bddc 100644
--- a/python/packages/orchestrations/agent_framework_orchestrations/_orchestration_request_info.py
+++ b/python/packages/orchestrations/agent_framework_orchestrations/_orchestration_request_info.py
@@ -122,21 +122,29 @@ def __init__(
         self,
         agent: SupportsAgentRun,
         context_mode: Literal["full", "last_agent", "custom"] | None = None,
+        *,
+        emit_data_events: bool = False,
     ) -> None:
         """Initialize the AgentApprovalExecutor.
 
         Args:
             agent: The agent protocol to use for generating responses.
             context_mode: The mode for providing context to the agent.
+            emit_data_events: Forwarded to the inner AgentExecutor.
         """
         self._context_mode: Literal["full", "last_agent", "custom"] | None = context_mode
         self._description = agent.description
+        self._emit_data_events = emit_data_events
 
         super().__init__(workflow=self._build_workflow(agent), id=resolve_agent_id(agent), propagate_request=True)
 
     def _build_workflow(self, agent: SupportsAgentRun) -> Workflow:
         """Build the internal workflow for the AgentApprovalExecutor."""
-        agent_executor = AgentExecutor(agent, context_mode=self._context_mode)
+        agent_executor = AgentExecutor(
+            agent,
+            context_mode=self._context_mode,
+            emit_data_events=self._emit_data_events,
+        )
         request_info_executor = AgentRequestInfoExecutor(id="agent_request_info_executor")
 
         return (
diff --git a/python/packages/orchestrations/agent_framework_orchestrations/_sequential.py b/python/packages/orchestrations/agent_framework_orchestrations/_sequential.py
index bda7f194ab..79fc9d28c5 100644
--- a/python/packages/orchestrations/agent_framework_orchestrations/_sequential.py
+++ b/python/packages/orchestrations/agent_framework_orchestrations/_sequential.py
@@ -2,46 +2,24 @@
 
 """Sequential builder for agent/executor workflows with shared conversation context.
 
-This module provides a high-level, agent-focused API to assemble a sequential
-workflow where:
-- Participants are provided as SupportsAgentRun or Executor instances via `participants=[...]`
-- A shared conversation context (list[Message]) is passed along the chain
-- Agents append their assistant messages to the context
-- Custom executors can transform or summarize and return a refined context
-- The workflow finishes with the final context produced by the last participant
-
-Typical wiring:
-    input -> _InputToConversation -> participant1 -> (agent? -> _ResponseToConversation) ->
-    ... -> participantN -> _EndWithConversation
-
-Notes:
-- Participants can mix SupportsAgentRun and Executor objects
-- Agents are auto-wrapped by WorkflowBuilder as AgentExecutor (unless already wrapped)
-- AgentExecutor produces AgentExecutorResponse; _ResponseToConversation converts this to list[Message]
-- Non-agent executors must define a handler that consumes `list[Message]` and sends back
-  the updated `list[Message]` via their workflow context
-
-Why include the small internal adapter executors?
-- Input normalization ("input-conversation"): ensures the workflow always starts with a
-  `list[Message]` regardless of whether callers pass a `str`, a single `Message`,
-  or a list. This keeps the first hop strongly typed and avoids boilerplate in participants.
-- Agent response adaptation ("to-conversation:<participant>"): agents (via AgentExecutor)
-  emit `AgentExecutorResponse`. The adapter converts that to a `list[Message]`
-  using `full_conversation` so original prompts aren't lost when chaining.
-- Result output ("end"): yields the final conversation list and the workflow becomes idle
-  giving a consistent terminal payload shape for both agents and custom executors.
-
-These adapters are first-class executors by design so they are type-checked at edges,
-observable (ExecutorInvoke/Completed events), and easily testable/reusable. Their IDs are
-deterministic and self-describing (for example, "to-conversation:writer") to reduce event-log
-confusion and to mirror how the concurrent builder uses explicit dispatcher/aggregator nodes.
+Participants (SupportsAgentRun or Executor instances) run in order, sharing a
+conversation along the chain. Agents append their assistant messages; custom executors
+transform and return a refined `list[Message]`.
+
+Wiring: input -> _InputToConversation -> participant1 -> ... -> participantN -> _EndWithConversation
+
+The workflow's final `output` event is either the last agent's `AgentResponse` (when the
+terminator is an agent) or the custom executor's `list[Message]`. With
+`intermediate_outputs=True`, intermediate agents emit `data` events (via
+`AgentExecutor.emit_data_events`) so consumers can observe them separately from the
+terminal answer.
 """
 
 import logging
 from collections.abc import Sequence
 from typing import Any, Literal
 
-from agent_framework import Message, SupportsAgentRun
+from agent_framework import AgentResponse, Message, SupportsAgentRun
 from agent_framework._workflows._agent_executor import (
     AgentExecutor,
     AgentExecutorResponse,
@@ -79,7 +57,20 @@ async def from_messages(self, messages: list[str | Message], ctx: WorkflowContex
 
 
 class _EndWithConversation(Executor):
-    """Terminates the workflow by emitting the final conversation context."""
+    """Graph terminator for the sequential workflow.
+
+    For custom-executor terminators, this emits the final `list[Message]` as an `output`
+    event (the executor's own contract). For agent terminators it is a passive sink: the
+    last `AgentExecutor` is itself registered as the workflow's output executor in
+    `SequentialBuilder.build()`, so its `yield_output` calls — a single `AgentResponse`
+    non-streaming, or per-chunk `AgentResponseUpdate` events streaming — become the
+    workflow's outputs directly.
+
+    Intermediate participants emit observation `data` events (via
+    `AgentExecutor.emit_data_events`) when `intermediate_outputs=True`; they never
+    emit `output` events because output_executors is restricted to the terminator
+    executor (the last agent or this node).
+    """
 
     @handler
     async def end_with_messages(
@@ -87,23 +78,24 @@ async def end_with_messages(
         conversation: list[Message],
         ctx: WorkflowContext[Any, list[Message]],
     ) -> None:
-        """Handler for ending with a list of Message.
-
-        This is used when the last participant is a custom executor.
-        """
+        """Yield the final conversation when the last participant is a custom executor."""
         await ctx.yield_output(list(conversation))
 
     @handler
     async def end_with_agent_executor_response(
         self,
         response: AgentExecutorResponse,
-        ctx: WorkflowContext[Any, list[Message] | None],
+        ctx: WorkflowContext[Any, AgentResponse],
     ) -> None:
-        """Handle case where last participant is an agent.
+        """Convert the agent-terminator response into a workflow output.
 
-        The agent is wrapped by AgentExecutor and emits AgentExecutorResponse.
+        When the last participant is a regular AgentExecutor (registered as the
+        output executor), this node is NOT in output_executors so the yield is
+        silently filtered — no duplicate output. When the last participant is an
+        AgentApprovalExecutor (or similar wrapper), this node IS the output
+        executor so the yield produces the workflow's terminal answer.
         """
-        await ctx.yield_output(response.full_conversation)
+        await ctx.yield_output(response.agent_response)
 
 
 class SequentialBuilder:
@@ -225,7 +217,13 @@ def with_request_info(
         return self
 
     def _resolve_participants(self) -> list[Executor]:
-        """Resolve participant instances into Executor objects."""
+        """Resolve participant instances into Executor objects.
+
+        Wraps `SupportsAgentRun` participants as `AgentExecutor`. When `intermediate_outputs=True`,
+        every wrapped agent except the final one is constructed with `emit_data_events=True`
+        so its responses surface as workflow `data` events without polluting the single `output`
+        event reserved for the final answer.
+        """
         if not self._participants:
             raise ValueError("No participants provided. Pass participants to the constructor.")
 
@@ -235,18 +233,32 @@ def _resolve_participants(self) -> list[Executor]:
             "last_agent" if self._chain_only_agent_responses else None
         )
 
+        last_idx = len(participants) - 1
         executors: list[Executor] = []
-        for p in participants:
+        for idx, p in enumerate(participants):
             if isinstance(p, Executor):
                 executors.append(p)
             elif isinstance(p, SupportsAgentRun):
+                emit_intermediate = self._intermediate_outputs and idx != last_idx
                 if self._request_info_enabled and (
                     not self._request_info_filter or resolve_agent_id(p) in self._request_info_filter
                 ):
                     # Handle request info enabled agents
-                    executors.append(AgentApprovalExecutor(p, context_mode=context_mode))
+                    executors.append(
+                        AgentApprovalExecutor(
+                            p,
+                            context_mode=context_mode,
+                            emit_data_events=emit_intermediate,
+                        )
+                    )
                 else:
-                    executors.append(AgentExecutor(p, context_mode=context_mode))
+                    executors.append(
+                        AgentExecutor(
+                            p,
+                            context_mode=context_mode,
+                            emit_data_events=emit_intermediate,
+                        )
+                    )
             else:
                 raise TypeError(f"Participants must be SupportsAgentRun or Executor instances. Got {type(p).__name__}.")
 
@@ -258,11 +270,12 @@ def build(self) -> Workflow:
         Wiring pattern:
         - _InputToConversation normalizes the initial input into list[Message]
         - For each participant in order:
-            - If Agent (or AgentExecutor): pass conversation to the agent, then optionally
-              route through a request info interceptor, then convert response to conversation
-              via _ResponseToConversation
-            - Else (custom Executor): pass conversation directly to the executor
-        - _EndWithConversation yields the final conversation and the workflow becomes idle
+            - Agent or AgentExecutor: receives the conversation/AgentExecutorResponse,
+              produces an AgentExecutorResponse forwarded downstream
+            - Custom Executor: receives list[Message] and forwards a list[Message]
+        - The workflow's `output_executor` is selected based on the last participant:
+            - Agent terminator: the last AgentExecutor itself (its yield_output is the answer)
+            - Custom-executor terminator: `_EndWithConversation` (yields the final list[Message])
         """
         # Internal nodes
         input_conv = _InputToConversation(id="input-conversation")
@@ -271,10 +284,15 @@ def build(self) -> Workflow:
         # Resolve participants and participant factories to executors
         participants: list[Executor] = self._resolve_participants()
 
+        last_executor = participants[-1]
+        output_executors: list[Executor | SupportsAgentRun] = [
+            last_executor if isinstance(last_executor, AgentExecutor) else end
+        ]
+
         builder = WorkflowBuilder(
             start_executor=input_conv,
             checkpoint_storage=self._checkpoint_storage,
-            output_executors=[end] if not self._intermediate_outputs else None,
+            output_executors=output_executors,
         )
 
         # Start of the chain is the input normalizer
diff --git a/python/packages/orchestrations/tests/test_concurrent.py b/python/packages/orchestrations/tests/test_concurrent.py
index 7d9a2bc534..5c879f6153 100644
--- a/python/packages/orchestrations/tests/test_concurrent.py
+++ b/python/packages/orchestrations/tests/test_concurrent.py
@@ -1,14 +1,21 @@
 # Copyright (c) Microsoft. All rights reserved.
 
-from typing import Any, cast
+from collections.abc import AsyncIterable, Awaitable
+from typing import Any, Literal, cast, overload
 
 import pytest
 from agent_framework import (
     AgentExecutorRequest,
     AgentExecutorResponse,
     AgentResponse,
+    AgentResponseUpdate,
+    AgentRunInputs,
+    AgentSession,
+    BaseAgent,
+    Content,
     Executor,
     Message,
+    ResponseStream,
     WorkflowContext,
     WorkflowRunState,
     handler,
@@ -49,36 +56,26 @@ def test_concurrent_builder_rejects_duplicate_executors() -> None:
         ConcurrentBuilder(participants=[a, b])
 
 
-async def test_concurrent_default_aggregator_emits_single_user_and_assistants() -> None:
-    # Three synthetic agent executors
+async def test_concurrent_default_aggregator_emits_assistants_only() -> None:
+    """Default aggregator yields a single AgentResponse with one assistant message per participant.
+
+    The user prompt is intentionally not included — that belongs in the input, not the answer.
+    """
     e1 = _FakeAgentExec("agentA", "Alpha")
     e2 = _FakeAgentExec("agentB", "Beta")
     e3 = _FakeAgentExec("agentC", "Gamma")
 
     wf = ConcurrentBuilder(participants=[e1, e2, e3]).build()
 
-    completed = False
-    output: list[Message] | None = None
-    async for ev in wf.run("prompt: hello world", stream=True):
-        if ev.type == "status" and ev.state == WorkflowRunState.IDLE:
-            completed = True
-        elif ev.type == "output":
-            output = cast(list[Message], ev.data)
-        if completed and output is not None:
-            break
-
-    assert completed
-    assert output is not None
-    messages: list[Message] = output
-
-    # Expect one user message + one assistant message per participant
-    assert len(messages) == 1 + 3
-    assert messages[0].role == "user"
-    assert "hello world" in messages[0].text
+    output_events = [ev for ev in await wf.run("prompt: hello world") if ev.type == "output"]
+    assert len(output_events) == 1
+    response = output_events[0].data
+    assert isinstance(response, AgentResponse)
 
-    assistant_texts = {m.text for m in messages[1:]}
-    assert assistant_texts == {"Alpha", "Beta", "Gamma"}
-    assert all(m.role == "assistant" for m in messages[1:])
+    # Exactly one assistant message per participant; no user prompt.
+    assert len(response.messages) == 3
+    assert all(m.role == "assistant" for m in response.messages)
+    assert {m.text for m in response.messages} == {"Alpha", "Beta", "Gamma"}
 
 
 async def test_concurrent_custom_aggregator_callback_is_used() -> None:
@@ -215,7 +212,7 @@ async def test_concurrent_checkpoint_resume_round_trip() -> None:
 
     wf = ConcurrentBuilder(participants=list(participants), checkpoint_storage=storage).build()
 
-    baseline_output: list[Message] | None = None
+    baseline_output: AgentResponse | None = None
     async for ev in wf.run("checkpoint concurrent", stream=True):
         if ev.type == "output":
             baseline_output = ev.data  # type: ignore[assignment]
@@ -236,7 +233,7 @@ async def test_concurrent_checkpoint_resume_round_trip() -> None:
     )
     wf_resume = ConcurrentBuilder(participants=list(resumed_participants), checkpoint_storage=storage).build()
 
-    resumed_output: list[Message] | None = None
+    resumed_output: AgentResponse | None = None
     async for ev in wf_resume.run(checkpoint_id=resume_checkpoint.checkpoint_id, stream=True):
         if ev.type == "output":
             resumed_output = ev.data  # type: ignore[assignment]
@@ -247,8 +244,8 @@ async def test_concurrent_checkpoint_resume_round_trip() -> None:
             break
 
     assert resumed_output is not None
-    assert [m.role for m in resumed_output] == [m.role for m in baseline_output]
-    assert [m.text for m in resumed_output] == [m.text for m in baseline_output]
+    assert [m.role for m in resumed_output.messages] == [m.role for m in baseline_output.messages]
+    assert [m.text for m in resumed_output.messages] == [m.text for m in baseline_output.messages]
 
 
 async def test_concurrent_checkpoint_runtime_only() -> None:
@@ -258,7 +255,7 @@ async def test_concurrent_checkpoint_runtime_only() -> None:
     agents = [_FakeAgentExec(id="agent1", reply_text="A1"), _FakeAgentExec(id="agent2", reply_text="A2")]
     wf = ConcurrentBuilder(participants=agents).build()
 
-    baseline_output: list[Message] | None = None
+    baseline_output: AgentResponse | None = None
     async for ev in wf.run("runtime checkpoint test", checkpoint_storage=storage, stream=True):
         if ev.type == "output":
             baseline_output = ev.data  # type: ignore[assignment]
@@ -278,7 +275,7 @@ async def test_concurrent_checkpoint_runtime_only() -> None:
     resumed_agents = [_FakeAgentExec(id="agent1", reply_text="A1"), _FakeAgentExec(id="agent2", reply_text="A2")]
     wf_resume = ConcurrentBuilder(participants=resumed_agents).build()
 
-    resumed_output: list[Message] | None = None
+    resumed_output: AgentResponse | None = None
     async for ev in wf_resume.run(
         checkpoint_id=resume_checkpoint.checkpoint_id, checkpoint_storage=storage, stream=True
     ):
@@ -291,7 +288,7 @@ async def test_concurrent_checkpoint_runtime_only() -> None:
             break
 
     assert resumed_output is not None
-    assert [m.role for m in resumed_output] == [m.role for m in baseline_output]
+    assert [m.role for m in resumed_output.messages] == [m.role for m in baseline_output.messages]
 
 
 async def test_concurrent_checkpoint_runtime_overrides_buildtime() -> None:
@@ -334,3 +331,84 @@ async def test_concurrent_builder_reusable_after_build_with_participants() -> No
 
     assert builder._participants[0] is e1  # type: ignore
     assert builder._participants[1] is e2  # type: ignore
+
+
+class _EchoAgent(BaseAgent):
+    """Simple agent that appends a single assistant message with its name."""
+
+    @overload
+    def run(
+        self,
+        messages: AgentRunInputs | None = ...,
+        *,
+        stream: Literal[False] = ...,
+        session: AgentSession | None = ...,
+        **kwargs: Any,
+    ) -> Awaitable[AgentResponse[Any]]: ...
+    @overload
+    def run(
+        self,
+        messages: AgentRunInputs | None = ...,
+        *,
+        stream: Literal[True],
+        session: AgentSession | None = ...,
+        **kwargs: Any,
+    ) -> ResponseStream[AgentResponseUpdate, AgentResponse[Any]]: ...
+
+    def run(
+        self,
+        messages: AgentRunInputs | None = None,
+        *,
+        stream: bool = False,
+        session: AgentSession | None = None,
+        **kwargs: Any,
+    ) -> Awaitable[AgentResponse[Any]] | ResponseStream[AgentResponseUpdate, AgentResponse[Any]]:
+        if stream:
+
+            async def _stream() -> AsyncIterable[AgentResponseUpdate]:
+                yield AgentResponseUpdate(contents=[Content.from_text(text=f"{self.name} reply")])
+
+            return ResponseStream(_stream(), finalizer=AgentResponse.from_updates)
+
+        async def _run() -> AgentResponse:
+            return AgentResponse(messages=[Message("assistant", [f"{self.name} reply"])])
+
+        return _run()
+
+
+async def test_concurrent_intermediate_outputs_emits_data_events() -> None:
+    """When intermediate_outputs=True, each participant emits a `data` event.
+
+    The single `output` event still carries the aggregated AgentResponse; per-participant
+    responses are emitted as `data` events so consumers can tell them apart.
+    """
+    a1 = _EchoAgent(id="agent1", name="A1")
+    a2 = _EchoAgent(id="agent2", name="A2")
+    a3 = _EchoAgent(id="agent3", name="A3")
+
+    wf = ConcurrentBuilder(participants=[a1, a2, a3], intermediate_outputs=True).build()
+
+    output_events = []
+    data_events = []
+    for ev in await wf.run("prompt: hello"):
+        if ev.type == "output":
+            output_events.append(ev)
+        elif ev.type == "data":
+            data_events.append(ev)
+
+    # One output event = the aggregated answer from the aggregator.
+    assert len(output_events) == 1
+    aggregated = output_events[0].data
+    assert isinstance(aggregated, AgentResponse)
+    assert len(aggregated.messages) == 3
+    assert all(m.role == "assistant" for m in aggregated.messages)
+
+    # Each participant emits a data event carrying its AgentResponse.
+    assert len(data_events) == 3
+    for dev in data_events:
+        assert isinstance(dev.data, AgentResponse)
+    data_texts = {dev.data.messages[0].text for dev in data_events}
+    assert data_texts == {"A1 reply", "A2 reply", "A3 reply"}
+    # Executor ids derive from the agent's name (resolve_agent_id behavior).
+    data_executor_ids = {dev.executor_id for dev in data_events}
+    assert data_executor_ids == {"A1", "A2", "A3"}
diff --git a/python/packages/orchestrations/tests/test_group_chat.py b/python/packages/orchestrations/tests/test_group_chat.py
index 2118de5ba7..0ae736cc4b 100644
--- a/python/packages/orchestrations/tests/test_group_chat.py
+++ b/python/packages/orchestrations/tests/test_group_chat.py
@@ -238,18 +238,18 @@ async def test_group_chat_builder_basic_flow() -> None:
         orchestrator_name="manager",
     ).build()
 
-    outputs: list[list[Message]] = []
+    outputs: list[AgentResponse] = []
     async for event in workflow.run("coordinate task", stream=True):
         if event.type == "output":
             data = event.data
-            if isinstance(data, list):
-                outputs.append(cast(list[Message], data))
+            if isinstance(data, AgentResponse):
+                outputs.append(data)
 
+    # Exactly one terminal `output` event = the orchestrator's completion AgentResponse.
     assert len(outputs) == 1
-    assert len(outputs[0]) >= 1
-    # Check that both agents contributed
-    authors = {msg.author_name for msg in outputs[0] if msg.author_name in ["alpha", "beta"]}
-    assert len(authors) == 2
+    assert outputs[0].messages
+    # The completion message is authored by the orchestrator.
+    assert outputs[0].messages[-1].author_name == "manager"
 
 
 async def test_group_chat_as_agent_accepts_conversation() -> None:
@@ -283,18 +283,18 @@ async def test_agent_manager_handles_concatenated_json_output() -> None:
         orchestrator_agent=manager,
     ).build()
 
-    outputs: list[list[Message]] = []
+    outputs: list[AgentResponse] = []
     async for event in workflow.run("coordinate task", stream=True):
         if event.type == "output":
             data = event.data
-            if isinstance(data, list):
-                outputs.append(cast(list[Message], data))
+            if isinstance(data, AgentResponse):
+                outputs.append(data)
 
     assert outputs
-    conversation = outputs[-1]
-    assert any(msg.author_name == "agent" and msg.text == "worker response" for msg in conversation)
-    assert conversation[-1].author_name == manager.name
-    assert conversation[-1].text == "concatenated manager final"
+    final_response = outputs[-1]
+    # Terminal AgentResponse contains only the orchestrator's completion message.
+    assert final_response.messages[-1].author_name == manager.name
+    assert final_response.messages[-1].text == "concatenated manager final"
 
 
 # Comprehensive tests for group chat functionality
@@ -400,19 +400,16 @@ def selector(state: GroupChatState) -> str:
             selection_func=selector,
         ).build()
 
-        outputs: list[list[Message]] = []
+        outputs: list[AgentResponse] = []
         async for event in workflow.run("test task", stream=True):
             if event.type == "output":
                 data = event.data
-                if isinstance(data, list):
-                    outputs.append(cast(list[Message], data))
-
-        # Should have terminated due to max_rounds, expect at least one output
-        assert len(outputs) >= 1
-        # The final message in the conversation should be about round limit
-        conversation = outputs[-1]
-        assert len(conversation) >= 1
-        final_output = conversation[-1]
+                if isinstance(data, AgentResponse):
+                    outputs.append(data)
+
+        # Exactly one terminal output event = orchestrator's max-rounds completion message.
+        assert len(outputs) == 1
+        final_output = outputs[0].messages[-1]
         assert "maximum number of rounds" in final_output.text.lower()
 
     async def test_termination_condition_halts_conversation(self) -> None:
@@ -431,22 +428,25 @@ def termination_condition(conversation: list[Message]) -> bool:
             participants=[agent],
             termination_condition=termination_condition,
             selection_func=selector,
+            intermediate_outputs=True,
         ).build()
 
-        outputs: list[list[Message]] = []
+        outputs: list[AgentResponse] = []
+        intermediate_updates: list[AgentResponseUpdate] = []
         async for event in workflow.run("test task", stream=True):
-            if event.type == "output":
-                data = event.data
-                if isinstance(data, list):
-                    outputs.append(cast(list[Message], data))
+            if event.type == "output" and isinstance(event.data, AgentResponse):
+                outputs.append(event.data)
+            elif event.type == "data" and isinstance(event.data, AgentResponseUpdate):
+                intermediate_updates.append(event.data)
 
         assert outputs, "Expected termination to yield output"
-        conversation = outputs[-1]
-        agent_replies = [msg for msg in conversation if msg.author_name == "agent" and msg.role == "assistant"]
-        assert len(agent_replies) == 2
-        final_output = conversation[-1]
-        # The orchestrator uses its ID as author_name by default
+        # Terminal output is the orchestrator's completion message only.
+        final_output = outputs[-1].messages[-1]
         assert "termination condition" in final_output.text.lower()
+        # Agent's intermediate replies surface as `data` events (per-update in streaming mode).
+        agent_updates = [u for u in intermediate_updates if u.author_name == "agent"]
+        # Each agent reply produces at least one update; expect 2 agent rounds before termination.
+        assert len(agent_updates) >= 2
 
     async def test_termination_condition_agent_manager_finalizes(self) -> None:
         """Test that termination condition with agent orchestrator produces default termination message."""
@@ -459,17 +459,17 @@ async def test_termination_condition_agent_manager_finalizes(self) -> None:
             orchestrator_agent=manager,
         ).build()
 
-        outputs: list[list[Message]] = []
+        outputs: list[AgentResponse] = []
         async for event in workflow.run("test task", stream=True):
             if event.type == "output":
                 data = event.data
-                if isinstance(data, list):
-                    outputs.append(cast(list[Message], data))
+                if isinstance(data, AgentResponse):
+                    outputs.append(data)
 
         assert outputs, "Expected termination to yield output"
-        conversation = outputs[-1]
-        assert conversation[-1].text == BaseGroupChatOrchestrator.TERMINATION_CONDITION_MET_MESSAGE
-        assert conversation[-1].author_name == manager.name
+        final_message = outputs[-1].messages[-1]
+        assert final_message.text == BaseGroupChatOrchestrator.TERMINATION_CONDITION_MET_MESSAGE
+        assert final_message.author_name == manager.name
 
     async def test_unknown_participant_error(self) -> None:
         """Test that unknown participant selection raises error."""
@@ -505,12 +505,12 @@ def selector(state: GroupChatState) -> str:
             selection_func=selector,
         ).build()
 
-        outputs: list[list[Message]] = []
+        outputs: list[AgentResponse] = []
         async for event in workflow.run("test task", stream=True):
             if event.type == "output":
                 data = event.data
-                if isinstance(data, list):
-                    outputs.append(cast(list[Message], data))
+                if isinstance(data, AgentResponse):
+                    outputs.append(data)
 
         assert len(outputs) == 1  # Should complete normally
 
@@ -546,12 +546,12 @@ def selector(state: GroupChatState) -> str:
 
         workflow = GroupChatBuilder(participants=[agent], max_rounds=1, selection_func=selector).build()
 
-        outputs: list[list[Message]] = []
+        outputs: list[AgentResponse] = []
         async for event in workflow.run("test string", stream=True):
             if event.type == "output":
                 data = event.data
-                if isinstance(data, list):
-                    outputs.append(cast(list[Message], data))
+                if isinstance(data, AgentResponse):
+                    outputs.append(data)
 
         assert len(outputs) == 1
 
@@ -569,12 +569,12 @@ def selector(state: GroupChatState) -> str:
 
         workflow = GroupChatBuilder(participants=[agent], max_rounds=1, selection_func=selector).build()
 
-        outputs: list[list[Message]] = []
+        outputs: list[AgentResponse] = []
         async for event in workflow.run(task_message, stream=True):
             if event.type == "output":
                 data = event.data
-                if isinstance(data, list):
-                    outputs.append(cast(list[Message], data))
+                if isinstance(data, AgentResponse):
+                    outputs.append(data)
 
         assert len(outputs) == 1
 
@@ -595,12 +595,12 @@ def selector(state: GroupChatState) -> str:
 
         workflow = GroupChatBuilder(participants=[agent], max_rounds=1, selection_func=selector).build()
 
-        outputs: list[list[Message]] = []
+        outputs: list[AgentResponse] = []
         async for event in workflow.run(conversation, stream=True):
             if event.type == "output":
                 data = event.data
-                if isinstance(data, list):
-                    outputs.append(cast(list[Message], data))
+                if isinstance(data, AgentResponse):
+                    outputs.append(data)
 
         assert len(outputs) == 1
 
@@ -625,19 +625,16 @@ def selector(state: GroupChatState) -> str:
             selection_func=selector,
         ).build()
 
-        outputs: list[list[Message]] = []
+        outputs: list[AgentResponse] = []
         async for event in workflow.run("test", stream=True):
             if event.type == "output":
                 data = event.data
-                if isinstance(data, list):
-                    outputs.append(cast(list[Message], data))
-
-        # Should have at least one output (the round limit message)
-        assert len(outputs) >= 1
-        # The last message in the conversation should be about round limit
-        conversation = outputs[-1]
-        assert len(conversation) >= 1
-        final_output = conversation[-1]
+                if isinstance(data, AgentResponse):
+                    outputs.append(data)
+
+        # Exactly one terminal output event = orchestrator's max-rounds completion message.
+        assert len(outputs) == 1
+        final_output = outputs[0].messages[-1]
         assert "maximum number of rounds" in final_output.text.lower()
 
     async def test_round_limit_in_ingest_participant_message(self) -> None:
@@ -658,19 +655,16 @@ def selector(state: GroupChatState) -> str:
             selection_func=selector,
         ).build()
 
-        outputs: list[list[Message]] = []
+        outputs: list[AgentResponse] = []
         async for event in workflow.run("test", stream=True):
             if event.type == "output":
                 data = event.data
-                if isinstance(data, list):
-                    outputs.append(cast(list[Message], data))
-
-        # Should have at least one output (the round limit message)
-        assert len(outputs) >= 1
-        # The last message in the conversation should be about round limit
-        conversation = outputs[-1]
-        assert len(conversation) >= 1
-        final_output = conversation[-1]
+                if isinstance(data, AgentResponse):
+                    outputs.append(data)
+
+        # Exactly one terminal output event = orchestrator's max-rounds completion message.
+        assert len(outputs) == 1
+        final_output = outputs[0].messages[-1]
         assert "maximum number of rounds" in final_output.text.lower()
 
 
@@ -684,10 +678,10 @@ async def test_group_chat_checkpoint_runtime_only() -> None:
 
     wf = GroupChatBuilder(participants=[agent_a, agent_b], max_rounds=2, selection_func=selector).build()
 
-    baseline_output: list[Message] | None = None
+    baseline_output: AgentResponse | None = None
     async for ev in wf.run("runtime checkpoint test", checkpoint_storage=storage, stream=True):
-        if ev.type == "output":
-            baseline_output = cast(list[Message], ev.data) if isinstance(ev.data, list) else None  # type: ignore
+        if ev.type == "output" and isinstance(ev.data, AgentResponse):
+            baseline_output = ev.data
         if ev.type == "status" and ev.state in (
             WorkflowRunState.IDLE,
             WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
@@ -720,10 +714,10 @@ async def test_group_chat_checkpoint_runtime_overrides_buildtime() -> None:
             checkpoint_storage=buildtime_storage,
             selection_func=selector,
         ).build()
-        baseline_output: list[Message] | None = None
+        baseline_output: AgentResponse | None = None
         async for ev in wf.run("override test", checkpoint_storage=runtime_storage, stream=True):
-            if ev.type == "output":
-                baseline_output = cast(list[Message], ev.data) if isinstance(ev.data, list) else None  # type: ignore
+            if ev.type == "output" and isinstance(ev.data, AgentResponse):
+                baseline_output = ev.data
             if ev.type == "status" and ev.state in (
                 WorkflowRunState.IDLE,
                 WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
@@ -975,13 +969,9 @@ def agent_factory() -> Agent:
 
     assert len(outputs) == 1
     # The DynamicManagerAgent terminates after second call with final_message
-    final_messages = outputs[0].data
-    assert isinstance(final_messages, list)
-    assert any(
-        msg.text == "dynamic manager final"
-        for msg in cast(list[Message], final_messages)
-        if msg.author_name == "dynamic_manager"
-    )
+    final_response = outputs[0].data
+    assert isinstance(final_response, AgentResponse)
+    assert any(msg.text == "dynamic manager final" for msg in final_response.messages)
 
 
 def test_group_chat_with_orchestrator_factory_returning_base_orchestrator():
diff --git a/python/packages/orchestrations/tests/test_handoff.py b/python/packages/orchestrations/tests/test_handoff.py
index a512d9b9df..33eed34406 100644
--- a/python/packages/orchestrations/tests/test_handoff.py
+++ b/python/packages/orchestrations/tests/test_handoff.py
@@ -9,6 +9,8 @@
 import pytest
 from agent_framework import (
     Agent,
+    AgentResponse,
+    AgentResponseUpdate,
     ChatResponse,
     ChatResponseUpdate,
     Content,
@@ -856,10 +858,15 @@ async def test_autonomous_mode_yields_output_without_user_request():
     outputs = [ev for ev in events if ev.type == "output"]
     assert outputs, "Autonomous mode should yield a workflow output"
 
-    final_conversation = outputs[-1].data
-    assert isinstance(final_conversation, list)
-    conversation_list = cast(list[Message], final_conversation)
-    assert any(msg.role == "assistant" and (msg.text or "").startswith("specialist reply") for msg in conversation_list)
+    # Per-agent activity surfaces as `output` events from each HandoffAgentExecutor as they
+    # speak. Handoff has no orchestrator that produces a separate "answer" — the conversation
+    # IS the result. In streaming mode payloads are AgentResponseUpdate; combined text should
+    # contain the specialist's reply.
+    payloads = [ev.data for ev in outputs if isinstance(ev.data, (AgentResponse, AgentResponseUpdate))]
+    combined = " ".join(
+        getattr(p, "text", None) or " ".join(m.text for m in getattr(p, "messages", [])) for p in payloads
+    )
+    assert "specialist reply" in combined
 
 
 async def test_autonomous_mode_resumes_user_input_on_turn_limit():
@@ -923,14 +930,10 @@ async def async_termination(conv: list[Message]) -> bool:
             stream=True, responses={requests[-1].request_id: [Message(role="user", contents=["Second user message"])]}
         )
     )
-    outputs = [ev for ev in events if ev.type == "output"]
-    assert len(outputs) == 1
-
-    final_conversation = outputs[0].data
-    assert isinstance(final_conversation, list)
-    final_conv_list = cast(list[Message], final_conversation)
-    user_messages = [msg for msg in final_conv_list if msg.role == "user"]
-    assert len(user_messages) == 2
+    # Resume run terminates without further agent activity once the second user message
+    # satisfies the termination condition. The workflow returns to idle cleanly.
+    idle_states = [ev for ev in events if ev.type == "status" and ev.state == WorkflowRunState.IDLE]
+    assert idle_states, "Workflow should become idle after termination"
     assert termination_call_count > 0
 
 
@@ -990,8 +993,9 @@ async def _get() -> ChatResponse:
 
     outputs = [event for event in events if event.type == "output"]
     assert outputs
-    conversation_outputs = [event for event in outputs if isinstance(event.data, list)]
-    assert len(conversation_outputs) == 1
+    # Per-agent activity surfaces as output events (AgentResponseUpdate in streaming mode).
+    agent_payloads = [event for event in outputs if isinstance(event.data, (AgentResponse, AgentResponseUpdate))]
+    assert len(agent_payloads) >= 1
 
 
 async def test_tool_choice_preserved_from_agent_config():
diff --git a/python/packages/orchestrations/tests/test_magentic.py b/python/packages/orchestrations/tests/test_magentic.py
index b87de8c6f4..96bfaeccfd 100644
--- a/python/packages/orchestrations/tests/test_magentic.py
+++ b/python/packages/orchestrations/tests/test_magentic.py
@@ -194,9 +194,9 @@ async def test_magentic_builder_returns_workflow_and_runs() -> None:
     orchestrator_event_count = 0
     async for event in workflow.run("compose summary", stream=True):
         if event.type == "output":
-            msg = event.data
-            if isinstance(msg, list):
-                outputs.extend(cast(list[Message], msg))
+            data = event.data
+            if isinstance(data, AgentResponse):
+                outputs.extend(data.messages)
         elif event.type == "magentic_orchestrator":
             orchestrator_event_count += 1
 
@@ -250,7 +250,7 @@ async def test_magentic_workflow_plan_review_approval_to_completion():
     assert isinstance(req_event.data, MagenticPlanReviewRequest)
 
     completed = False
-    output: list[Message] | None = None
+    output: AgentResponse | None = None
     async for ev in wf.run(stream=True, responses={req_event.request_id: req_event.data.approve()}):
         if ev.type == "status" and ev.state == WorkflowRunState.IDLE:
             completed = True
@@ -261,8 +261,8 @@ async def test_magentic_workflow_plan_review_approval_to_completion():
 
     assert completed
     assert output is not None
-    assert isinstance(output, list)
-    assert all(isinstance(msg, Message) for msg in output)
+    assert isinstance(output, AgentResponse)
+    assert all(isinstance(msg, Message) for msg in output.messages)
 
 
 async def test_magentic_plan_review_with_revise():
@@ -337,10 +337,10 @@ async def test_magentic_orchestrator_round_limit_produces_partial_result():
     output_event = next((e for e in events if e.type == "output"), None)
     assert output_event is not None
     data = output_event.data
-    assert isinstance(data, list)
-    assert len(data) > 0  # type: ignore
-    assert data[-1].role == "assistant"  # type: ignore
-    assert all(isinstance(msg, Message) for msg in data)  # type: ignore
+    assert isinstance(data, AgentResponse)
+    assert len(data.messages) > 0
+    assert data.messages[-1].role == "assistant"
+    assert all(isinstance(msg, Message) for msg in data.messages)
 
 
 async def test_magentic_checkpoint_resume_round_trip():
@@ -576,12 +576,11 @@ async def _collect_agent_responses_setup(participant: SupportsAgentRun) -> list[
 
     wf = MagenticBuilder(participants=[participant], intermediate_outputs=True, manager=InvokeOnceManager()).build()
 
-    # Run a bounded stream to allow one invoke and then completion
-    events: list[WorkflowEvent] = []
-    async for ev in wf.run("task", stream=True):  # plan review disabled
-        events.append(ev)
-        # Capture streaming updates (type="output" with AgentResponseUpdate data)
-        if ev.type == "output" and isinstance(ev.data, AgentResponseUpdate):
+    # With intermediate_outputs=True, participant updates surface as `data` events
+    # carrying AgentResponseUpdate; the orchestrator's terminal AgentResponse comes via
+    # an `output` event.
+    async for ev in wf.run("task", stream=True):
+        if ev.type == "data" and isinstance(ev.data, AgentResponseUpdate):
             captured.append(
                 Message(
                     role=ev.data.role or "assistant",
@@ -589,7 +588,6 @@ async def _collect_agent_responses_setup(participant: SupportsAgentRun) -> list[
                     author_name=ev.data.author_name,
                 )
             )
-        # Break on final AgentResponse output
         elif ev.type == "output" and isinstance(ev.data, AgentResponse):
             break
 
@@ -753,11 +751,12 @@ async def test_magentic_stall_and_reset_reach_limits():
     assert idle_status is not None
     output_event = next((e for e in events if e.type == "output"), None)
     assert output_event is not None
-    assert isinstance(output_event.data, list)
-    assert all(isinstance(msg, Message) for msg in output_event.data)  # type: ignore
-    assert len(output_event.data) > 0  # type: ignore
-    assert output_event.data[-1].text is not None  # type: ignore
-    assert output_event.data[-1].text == "Workflow terminated due to reaching maximum reset count."  # type: ignore
+    assert isinstance(output_event.data, AgentResponse)
+    msgs = output_event.data.messages
+    assert all(isinstance(msg, Message) for msg in msgs)
+    assert len(msgs) > 0
+    assert msgs[-1].text is not None
+    assert msgs[-1].text == "Workflow terminated due to reaching maximum reset count."
 
 
 async def test_magentic_checkpoint_runtime_only() -> None:
diff --git a/python/packages/orchestrations/tests/test_sequential.py b/python/packages/orchestrations/tests/test_sequential.py
index 0f000ef254..1dd5bef45a 100644
--- a/python/packages/orchestrations/tests/test_sequential.py
+++ b/python/packages/orchestrations/tests/test_sequential.py
@@ -98,32 +98,126 @@ def test_sequential_builder_validation_rejects_invalid_executor() -> None:
         SequentialBuilder(participants=[_EchoAgent(id="agent1", name="A1"), _InvalidExecutor(id="invalid")]).build()
 
 
-async def test_sequential_agents_append_to_context() -> None:
+async def test_sequential_streaming_yields_only_last_agent_updates() -> None:
+    """Streaming mode surfaces only the last agent's AgentResponseUpdate chunks as outputs.
+
+    Intermediate agents do NOT emit `output` events when intermediate_outputs=False (default);
+    only the last agent (the workflow's output_executor) emits chunks of the final answer.
+    """
     a1 = _EchoAgent(id="agent1", name="A1")
     a2 = _EchoAgent(id="agent2", name="A2")
 
     wf = SequentialBuilder(participants=[a1, a2]).build()
 
     completed = False
-    output: list[Message] | None = None
+    update_events: list[AgentResponseUpdate] = []
     async for ev in wf.run("hello sequential", stream=True):
         if ev.type == "status" and ev.state == WorkflowRunState.IDLE:
             completed = True
         elif ev.type == "output":
-            output = ev.data  # type: ignore[assignment]
-        if completed and output is not None:
+            update_events.append(ev.data)  # type: ignore[arg-type]
+        if completed:
             break
 
     assert completed
-    assert output is not None
-    assert isinstance(output, list)
-    msgs: list[Message] = output
-    assert len(msgs) == 3
-    assert msgs[0].role == "user" and "hello sequential" in msgs[0].text
-    assert msgs[1].role == "assistant" and (msgs[1].author_name == "A1" or True)
-    assert msgs[2].role == "assistant" and (msgs[2].author_name == "A2" or True)
-    assert "A1 reply" in msgs[1].text
-    assert "A2 reply" in msgs[2].text
+    # Only the last agent's streaming chunks surface as `output` events.
+    assert update_events, "Expected at least one streaming update from the last agent"
+    for upd in update_events:
+        assert isinstance(upd, AgentResponseUpdate)
+    combined_text = "".join(u.text for u in update_events if hasattr(u, "text"))
+    assert "A2 reply" in combined_text
+    assert "A1 reply" not in combined_text
+
+
+async def test_sequential_non_streaming_yields_only_last_agent_response() -> None:
+    """Non-streaming mode emits a single `output` event with the last agent's AgentResponse."""
+    a1 = _EchoAgent(id="agent1", name="A1")
+    a2 = _EchoAgent(id="agent2", name="A2")
+
+    wf = SequentialBuilder(participants=[a1, a2]).build()
+
+    output_events = [ev for ev in await wf.run("hello sequential") if ev.type == "output"]
+    assert len(output_events) == 1
+    response = output_events[0].data
+    assert isinstance(response, AgentResponse)
+    assert all(m.role == "assistant" for m in response.messages)
+    combined = " ".join(m.text for m in response.messages)
+    assert "A2 reply" in combined
+    assert "A1 reply" not in combined
+
+
+async def test_sequential_intermediate_outputs_emits_data_events() -> None:
+    """When intermediate_outputs=True, intermediate agents surface as `data` events.
+
+    The single `output` event still carries the last agent's AgentResponse; intermediate
+    agents are emitted as `data` events (not output events) so consumers can clearly tell
+    them apart.
+    """
+    a1 = _EchoAgent(id="agent1", name="A1")
+    a2 = _EchoAgent(id="agent2", name="A2")
+
+    wf = SequentialBuilder(participants=[a1, a2], intermediate_outputs=True).build()
+
+    output_events = []
+    data_events = []
+    for ev in await wf.run("hello"):
+        if ev.type == "output":
+            output_events.append(ev)
+        elif ev.type == "data":
+            data_events.append(ev)
+
+    # One output event = the final answer (last agent).
+    assert len(output_events) == 1
+    final = output_events[0].data
+    assert isinstance(final, AgentResponse)
+    assert "A2 reply" in " ".join(m.text for m in final.messages)
+
+    # Intermediate agents emit data events (not output events).
+    assert len(data_events) == 1
+    intermediate = data_events[0].data
+    assert isinstance(intermediate, AgentResponse)
+    assert "A1 reply" in " ".join(m.text for m in intermediate.messages)
+    # Executor id derives from the agent's name (resolve_agent_id behavior).
+    assert data_events[0].executor_id == "A1"
+
+
+async def test_sequential_as_agent_returns_only_last_agent_response() -> None:
+    """`workflow.as_agent().run(prompt)` returns ONLY the last agent's messages — not the user
+    input or earlier agents' replies. This is the core fix for the orchestration-as-agent
+    output contract."""
+    a1 = _EchoAgent(id="agent1", name="A1")
+    a2 = _EchoAgent(id="agent2", name="A2")
+
+    agent = SequentialBuilder(participants=[a1, a2]).build().as_agent()
+    response = await agent.run("hello as_agent")
+
+    assert isinstance(response, AgentResponse)
+    # Only the last agent's reply — no user prompt, no agent1 messages.
+    combined = " ".join(m.text for m in response.messages)
+    assert "A2 reply" in combined
+    assert "A1 reply" not in combined
+    assert "hello as_agent" not in combined
+
+
+async def test_sequential_as_agent_with_intermediate_outputs_includes_chain() -> None:
+    """With `intermediate_outputs=True`, `as_agent()` surfaces intermediate agent responses
+    (rewritten as `text_reasoning` content) followed by the final answer (`text` content)."""
+    a1 = _EchoAgent(id="agent1", name="A1")
+    a2 = _EchoAgent(id="agent2", name="A2")
+
+    agent = SequentialBuilder(participants=[a1, a2], intermediate_outputs=True).build().as_agent()
+    response = await agent.run("hello as_agent")
+
+    assert isinstance(response, AgentResponse)
+
+    reasoning_text = " ".join(c.text or "" for m in response.messages for c in m.contents if c.type == "text_reasoning")
+    answer_text = " ".join(c.text or "" for m in response.messages for c in m.contents if c.type == "text")
+    assert "A1 reply" in reasoning_text, "Intermediate writer reply should arrive as reasoning content"
+    assert "A2 reply" in answer_text, "Terminal reviewer reply should arrive as text content"
+    assert "A1 reply" not in answer_text, "Intermediate content should not appear as final text"
+    # Final agent's reply should appear last in the message ordering.
+    last_msg_text = " ".join(c.text or "" for c in response.messages[-1].contents if c.type == "text")
+    assert "A2 reply" in last_msg_text
 
 
 async def test_sequential_with_custom_executor_summary() -> None:
@@ -158,14 +252,14 @@ async def test_sequential_checkpoint_resume_round_trip() -> None:
     initial_agents = (_EchoAgent(id="agent1", name="A1"), _EchoAgent(id="agent2", name="A2"))
     wf = SequentialBuilder(participants=list(initial_agents), checkpoint_storage=storage).build()
 
-    baseline_output: list[Message] | None = None
+    baseline_updates: list[AgentResponseUpdate] = []
     async for ev in wf.run("checkpoint sequential", stream=True):
         if ev.type == "output":
-            baseline_output = ev.data  # type: ignore[assignment]
+            baseline_updates.append(ev.data)  # type: ignore[arg-type]
         if ev.type == "status" and ev.state == WorkflowRunState.IDLE:
             break
 
-    assert baseline_output is not None
+    assert baseline_updates
 
     checkpoints = await storage.list_checkpoints(workflow_name=wf.name)
     assert checkpoints
@@ -175,19 +269,20 @@ async def test_sequential_checkpoint_resume_round_trip() -> None:
     resumed_agents = (_EchoAgent(id="agent1", name="A1"), _EchoAgent(id="agent2", name="A2"))
     wf_resume = SequentialBuilder(participants=list(resumed_agents), checkpoint_storage=storage).build()
 
-    resumed_output: list[Message] | None = None
+    resumed_updates: list[AgentResponseUpdate] = []
     async for ev in wf_resume.run(checkpoint_id=resume_checkpoint.checkpoint_id, stream=True):
         if ev.type == "output":
-            resumed_output = ev.data  # type: ignore[assignment]
+            resumed_updates.append(ev.data)  # type: ignore[arg-type]
         if ev.type == "status" and ev.state in (
             WorkflowRunState.IDLE,
             WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
         ):
             break
 
-    assert resumed_output is not None
-    assert [m.role for m in resumed_output] == [m.role for m in baseline_output]
-    assert [m.text for m in resumed_output] == [m.text for m in baseline_output]
+    assert resumed_updates
+    baseline_text = "".join(u.text for u in baseline_updates if hasattr(u, "text"))
+    resumed_text = "".join(u.text for u in resumed_updates if hasattr(u, "text"))
+    assert baseline_text == resumed_text
 
 
 async def test_sequential_checkpoint_runtime_only() -> None:
@@ -197,14 +292,14 @@ async def test_sequential_checkpoint_runtime_only() -> None:
     agents = (_EchoAgent(id="agent1", name="A1"), _EchoAgent(id="agent2", name="A2"))
     wf = SequentialBuilder(participants=list(agents)).build()
 
-    baseline_output: list[Message] | None = None
+    baseline_updates: list[AgentResponseUpdate] = []
     async for ev in wf.run("runtime checkpoint test", checkpoint_storage=storage, stream=True):
         if ev.type == "output":
-            baseline_output = ev.data  # type: ignore[assignment]
+            baseline_updates.append(ev.data)  # type: ignore[arg-type]
         if ev.type == "status" and ev.state == WorkflowRunState.IDLE:
             break
 
-    assert baseline_output is not None
+    assert baseline_updates
 
     checkpoints = await storage.list_checkpoints(workflow_name=wf.name)
     assert checkpoints
@@ -214,21 +309,22 @@ async def test_sequential_checkpoint_runtime_only() -> None:
     resumed_agents = (_EchoAgent(id="agent1", name="A1"), _EchoAgent(id="agent2", name="A2"))
     wf_resume = SequentialBuilder(participants=list(resumed_agents)).build()
 
-    resumed_output: list[Message] | None = None
+    resumed_updates: list[AgentResponseUpdate] = []
     async for ev in wf_resume.run(
         checkpoint_id=resume_checkpoint.checkpoint_id, checkpoint_storage=storage, stream=True
     ):
         if ev.type == "output":
-            resumed_output = ev.data  # type: ignore[assignment]
+            resumed_updates.append(ev.data)  # type: ignore[arg-type]
         if ev.type == "status" and ev.state in (
             WorkflowRunState.IDLE,
             WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
         ):
             break
 
-    assert resumed_output is not None
-    assert [m.role for m in resumed_output] == [m.role for m in baseline_output]
-    assert [m.text for m in resumed_output] == [m.text for m in baseline_output]
+    assert resumed_updates
+    baseline_text = "".join(u.text for u in baseline_updates if hasattr(u, "text"))
+    resumed_text = "".join(u.text for u in resumed_updates if hasattr(u, "text"))
+    assert baseline_text == resumed_text
 
 
 async def test_sequential_checkpoint_runtime_overrides_buildtime() -> None:
@@ -390,3 +486,102 @@ async def test_chain_only_agent_responses_three_agents() -> None:
     # a3 should see only A2's reply
     assert len(a3.last_messages) == 1
     assert a3.last_messages[0].role == "assistant" and "A2 reply" in (a3.last_messages[0].text or "")
+
+
+# ---------------------------------------------------------------------------
+# with_request_info tests
+# ---------------------------------------------------------------------------
+
+
+async def test_sequential_request_info_last_participant_emits_output() -> None:
+    """When the last participant is wrapped via with_request_info(), the workflow
+    still emits a terminal output event after approval.
+
+    This exercises the _EndWithConversation.end_with_agent_executor_response path
+    that converts the AgentApprovalExecutor's forwarded AgentExecutorResponse into
+    the workflow's final AgentResponse output.
+    """
+    from agent_framework_orchestrations._orchestration_request_info import AgentRequestInfoResponse
+
+    a1 = _EchoAgent(id="agent1", name="A1")
+    a2 = _EchoAgent(id="agent2", name="A2")
+
+    wf = SequentialBuilder(participants=[a1, a2]).with_request_info().build()
+
+    # First run: collect request_info events for both agents
+    request_events: list[Any] = []
+    async for ev in wf.run("hello with approval", stream=True):
+        if ev.type == "request_info" and isinstance(ev.data, AgentExecutorResponse):
+            request_events.append(ev)
+
+    # Approve each agent in sequence until the workflow completes
+    while request_events:
+        responses = {req.request_id: AgentRequestInfoResponse.approve() for req in request_events}
+        request_events = []
+        output_events: list[Any] = []
+        async for ev in wf.run(stream=True, responses=responses):
+            if ev.type == "request_info" and isinstance(ev.data, AgentExecutorResponse):
+                request_events.append(ev)
+            elif ev.type == "output":
+                output_events.append(ev)
+
+    # The workflow must produce a terminal output with the last agent's response.
+    assert len(output_events) == 1
+    response = output_events[0].data
+    assert isinstance(response, AgentResponse)
+    assert any("A2 reply" in m.text for m in response.messages)
+
+
+async def test_sequential_request_info_with_intermediate_outputs_emits_data_events() -> None:
+    """With both with_request_info() and intermediate_outputs=True, intermediate
+    agents' responses are surfaced as data events while the final output is an
+    AgentResponse from the last agent.
+
+    This verifies that WorkflowExecutor correctly forwards data events from the
+    inner AgentExecutor through the AgentApprovalExecutor wrapper.
+    """
+    from agent_framework_orchestrations._orchestration_request_info import AgentRequestInfoResponse
+
+    a1 = _EchoAgent(id="agent1", name="A1")
+    a2 = _EchoAgent(id="agent2", name="A2")
+
+    wf = SequentialBuilder(participants=[a1, a2], intermediate_outputs=True).with_request_info().build()
+
+    # Run and approve all request_info events until the workflow completes
+    all_data_events: list[Any] = []
+    all_output_events: list[Any] = []
+    request_events: list[Any] = []
+
+    async for ev in wf.run("hello intermediate", stream=True):
+        if ev.type == "request_info" and isinstance(ev.data, AgentExecutorResponse):
+            request_events.append(ev)
+        elif ev.type == "data":
+            all_data_events.append(ev)
+        elif ev.type == "output":
+            all_output_events.append(ev)
+
+    while request_events:
+        responses = {req.request_id: AgentRequestInfoResponse.approve() for req in request_events}
+        request_events = []
+        async for ev in wf.run(stream=True, responses=responses):
+            if ev.type == "request_info" and isinstance(ev.data, AgentExecutorResponse):
+                request_events.append(ev)
+            elif ev.type == "data":
+                all_data_events.append(ev)
+            elif ev.type == "output":
+                all_output_events.append(ev)
+
+    # The first (intermediate) agent should emit a data event.
+    assert len(all_data_events) >= 1
+    intermediate_texts = set()
+    for dev in all_data_events:
+        if isinstance(dev.data, AgentResponse):
+            for m in dev.data.messages:
+                intermediate_texts.add(m.text)
+    assert "A1 reply" in intermediate_texts
+
+    # The final output should contain the last agent's response.
+    assert len(all_output_events) >= 1
+    final = all_output_events[-1].data
+    assert isinstance(final, AgentResponse)
+    assert any("A2 reply" in m.text for m in final.messages)
diff --git a/python/samples/02-agents/conversations/file_history_provider.py b/python/samples/02-agents/conversations/file_history_provider.py
index 04a87f8224..20735ffd17 100644
--- a/python/samples/02-agents/conversations/file_history_provider.py
+++ b/python/samples/02-agents/conversations/file_history_provider.py
@@ -21,7 +21,7 @@
 from pydantic import Field
 
 try:
-    import orjson
+    import orjson  # pyright: ignore[reportMissingImports]
 except ImportError:
     orjson = None
 
diff --git a/python/samples/02-agents/conversations/file_history_provider_conversation_persistence.py b/python/samples/02-agents/conversations/file_history_provider_conversation_persistence.py
index 70c5d7e8e8..693501b0f9 100644
--- a/python/samples/02-agents/conversations/file_history_provider_conversation_persistence.py
+++ b/python/samples/02-agents/conversations/file_history_provider_conversation_persistence.py
@@ -22,7 +22,7 @@
 from pydantic import Field
 
 try:
-    import orjson
+    import orjson  # pyright: ignore[reportMissingImports]
 except ImportError:
     orjson = None
 
diff --git a/python/samples/03-workflows/agents/sequential_workflow_as_agent.py b/python/samples/03-workflows/agents/sequential_workflow_as_agent.py
index 52de975173..120bd448aa 100644
--- a/python/samples/03-workflows/agents/sequential_workflow_as_agent.py
+++ b/python/samples/03-workflows/agents/sequential_workflow_as_agent.py
@@ -26,8 +26,8 @@
   You can safely ignore them when focusing on agent progress.
 
 Prerequisites:
-- FOUNDRY_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
-- FOUNDRY_MODEL must be set to your Azure OpenAI model deployment name.
+- FOUNDRY_PROJECT_ENDPOINT must be set to the Azure Foundry project endpoint.
+- FOUNDRY_MODEL must be set to the model name for the Foundry chat client.
 """
 
 
@@ -68,28 +68,18 @@ async def main() -> None:
     """
     Sample Output:
 
-    ===== Final Conversation =====
+    ===== Conversation =====
     ------------------------------------------------------------
-    01 [user]
-    Write a tagline for a budget-friendly eBike.
-    ------------------------------------------------------------
-    02 [writer]
-    Ride farther, spend less—your affordable eBike adventure starts here.
-    ------------------------------------------------------------
-    03 [reviewer]
-    This tagline clearly communicates affordability and the benefit of extended travel, making it
-    appealing to budget-conscious consumers. It has a friendly and motivating tone, though it could
-    be slightly shorter for more punch. Overall, a strong and effective suggestion!
-
-    ===== as_agent() Conversation =====
-    ------------------------------------------------------------
-    01 [writer]
-    Go electric, save big—your affordable ride awaits!
-    ------------------------------------------------------------
-    02 [reviewer]
+    01 [reviewer]
     Catchy and straightforward! The tagline clearly emphasizes both the electric aspect and the affordability of the
     eBike. It's inviting and actionable. For even more impact, consider making it slightly shorter:
     "Go electric, save big." Overall, this is an effective and appealing suggestion for a budget-friendly eBike.
+
+    Note:
+    `workflow.as_agent()` returns ONLY the final agent's response (the "answer") — the prior agents' work
+    is not included in the response. To observe intermediate agents while running as an agent, build with
+    `SequentialBuilder(participants=[...], intermediate_outputs=True)`; the intermediate replies are then
+    surfaced as `data` events and merged into the AgentResponse.
     """