Improve the handling of intermediate outputs for workflows and orchestrations

python/packages/core/agent_framework/_workflows/_agent.py

@@ -32,6 +32,7 @@ from .._types import (
 from ..exceptions import AgentInvalidRequestException, AgentInvalidResponseException
 from ._checkpoint import CheckpointStorage
 from ._events import (
+    _LIFECYCLE_EVENT_TYPES,
     WorkflowEvent,
 )
 from ._message_utils import normalize_messages_input
@@ -48,6 +49,26 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
+def _to_text_reasoning(contents: Sequence[Content]) -> list[Content]:
+    """Rewrite text items as text_reasoning; non-text content passes through unchanged."""
+    rewritten: list[Content] = []
+    for c in contents:
+        if c.type == "text":
+            rewritten.append(
+                Content.from_text_reasoning(
+                    id=c.id,
+                    text=c.text,
+                    protected_data=c.protected_data,
+                    annotations=c.annotations,
+                    additional_properties=c.additional_properties,
+                    raw_representation=c.raw_representation,
+                )
+            )
+        else:
+            rewritten.append(c)
+    return rewritten
+
+
 class WorkflowAgent(BaseAgent):
     """An `Agent` subclass that wraps a workflow and exposes it as an agent."""
 
@@ -300,7 +321,7 @@ class WorkflowAgent(BaseAgent):
             function_invocation_kwargs=function_invocation_kwargs,
             client_kwargs=client_kwargs,
         ):
-            if event.type == "output" or event.type == "request_info":
+            if event.type not in _LIFECYCLE_EVENT_TYPES:
                 output_events.append(event)
 
         result = self._convert_workflow_events_to_agent_response(response_id, output_events)
@@ -514,7 +535,14 @@ class WorkflowAgent(BaseAgent):
         response_id: str,
         output_events: list[WorkflowEvent[Any]],
     ) -> AgentResponse:
-        """Convert a list of workflow output events to an AgentResponse."""
+        """Convert a list of workflow events to an AgentResponse.
+
+        Terminal events (``type='output'``) keep their text content as ``text``.
+        Intermediate events (``type='intermediate'`` and the legacy/orchestration
+        variants) have their text content rewritten to ``text_reasoning`` so
+        ``response.text`` returns terminal-only by virtue of the existing
+        ``Message.text`` filter.
+        """
         messages: list[Message] = []
         raw_representations: list[object] = []
         merged_usage: UsageDetails | None = None
@@ -535,18 +563,31 @@ class WorkflowAgent(BaseAgent):
                 raw_representations.append(output_event)
             else:
                 data = output_event.data
+                # Anything that isn't `output` is intermediate — this branch only sees
+                # events that already passed the lifecycle filter and weren't request_info.
+                is_intermediate = output_event.type != "output"
+
+                def _mark_msg(msg: Message) -> Message:
+                    return Message(
+                        contents=_to_text_reasoning(msg.contents),
+                        role=msg.role,
+                        author_name=msg.author_name,
+                        message_id=msg.message_id,
+                        additional_properties=msg.additional_properties,
+                        raw_representation=msg.raw_representation,
+                    )
 
                 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
-                    # responses.
+                    # AgentResponseUpdate in non-streaming mode would break message ordering
+                    # if interleaved with non-streaming AgentResponses.
                     raise AgentInvalidRequestException(
                         "Output event with AgentResponseUpdate data cannot be emitted in non-streaming mode. "
                         "Please ensure executors emit AgentResponse for non-streaming workflows."
                     )
 
                 if isinstance(data, AgentResponse):
-                    messages.extend(data.messages)
+                    inner_msgs = [_mark_msg(m) for m in data.messages] if is_intermediate else list(data.messages)
+                    messages.extend(inner_msgs)
                     raw_representations.append(data.raw_representation)
                     merged_usage = add_usage_details(merged_usage, data.usage_details)
                     latest_created_at = (
@@ -557,16 +598,19 @@ class WorkflowAgent(BaseAgent):
                         else latest_created_at
                     )
                 elif isinstance(data, Message):
-                    messages.append(data)
+                    messages.append(_mark_msg(data) if is_intermediate else data)
                     raw_representations.append(data.raw_representation)
                 elif is_instance_of(data, list[Message]):
                     chat_messages = cast(list[Message], data)
-                    messages.extend(chat_messages)
+                    inner_msgs = [_mark_msg(m) for m in chat_messages] if is_intermediate else list(chat_messages)
+                    messages.extend(inner_msgs)
                     raw_representations.append(data)
                 else:
                     contents = self._extract_contents(data)
                     if not contents:
                         continue
+                    if is_intermediate:
+                        contents = _to_text_reasoning(contents)
 
                     messages.append(
                         Message(
@@ -626,25 +670,35 @@ class WorkflowAgent(BaseAgent):
     ) -> 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.
+        Forwarding rule:
 
-        For 'output' events, AgentExecutor yields AgentResponseUpdate for streaming updates
-        via ctx.yield_output(). This method converts those to agent response updates.
-
-        Returns:
-            A list of AgentResponseUpdate objects. Empty list if the event is not relevant.
+        - ``type='output'`` — terminal user-facing emission. Forwarded as-is.
+        - ``type='request_info'`` — request-info translation (unchanged).
+        - Any other typed event (``intermediate``, the deprecated ``data``,
+          orchestration-specific types) — forwarded as intermediate. Text payloads
+          are rewritten to ``text_reasoning`` content; non-text content types
+          (function_call, function_result, data, uri, …) pass through unchanged
+          since their ``Content.type`` already discriminates them.
+        - Lifecycle / diagnostic events (``started``/``status``/``failed``/
+          ``warning``/``error``/``superstep_*``/``executor_*``) are dropped.
         """
-        if event.type == "output":
+        if event.type in _LIFECYCLE_EVENT_TYPES:
+            return []
+
+        if event.type != "request_info":
             data = event.data
             executor_id = event.executor_id
+            is_intermediate = event.type != "output"
+
+            def _maybe_mark(contents: list[Content]) -> list[Content]:
+                return _to_text_reasoning(contents) if is_intermediate else contents
 
             if isinstance(data, AgentResponseUpdate):
                 # Construct a fresh AgentResponseUpdate so we don't mutate a payload
                 # that AgentExecutor still holds a reference to in its `updates` list.
                 return [
                     AgentResponseUpdate(
-                        contents=list(data.contents),
+                        contents=_maybe_mark(list(data.contents)),
                         role=data.role,
                         author_name=data.author_name or executor_id,
                         response_id=data.response_id,
@@ -659,7 +713,7 @@ class WorkflowAgent(BaseAgent):
                 for msg in data.messages:
                     updates.append(
                         AgentResponseUpdate(
-                            contents=list(msg.contents),
+                            contents=_maybe_mark(list(msg.contents)),
                             role=msg.role,
                             author_name=msg.author_name or executor_id,
                             response_id=data.response_id or response_id,
@@ -673,7 +727,7 @@ class WorkflowAgent(BaseAgent):
             if isinstance(data, Message):
                 return [
                     AgentResponseUpdate(
-                        contents=list(data.contents),
+                        contents=_maybe_mark(list(data.contents)),
                         role=data.role,
                         author_name=data.author_name or executor_id,
                         response_id=response_id,
@@ -689,7 +743,7 @@ class WorkflowAgent(BaseAgent):
                 for msg in chat_messages:
                     updates.append(
                         AgentResponseUpdate(
-                            contents=list(msg.contents),
+                            contents=_maybe_mark(list(msg.contents)),
                             role=msg.role,
                             author_name=msg.author_name or executor_id,
                             response_id=response_id,
@@ -702,6 +756,10 @@ class WorkflowAgent(BaseAgent):
             contents = self._extract_contents(data)
             if not contents:
                 return []
+            if is_intermediate:
+                # _extract_contents returned text content for fallback str()-coercion of
+                # arbitrary payloads; reroute to text_reasoning for intermediate events.
+                contents = _maybe_mark(contents)
             return [
                 AgentResponseUpdate(
                     contents=contents,

python/packages/core/agent_framework/_workflows/_events.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import builtins
 import sys
 import traceback as _traceback
+import warnings
 from collections.abc import Iterator
 from contextlib import contextmanager
 from contextvars import ContextVar
@@ -106,8 +107,9 @@ WorkflowEventType = Literal[
     "status",  # Workflow state changed (use .state)
     "failed",  # Workflow terminated with error (use .details)
     # Data events
-    "output",  # Executor yielded final output (use .executor_id, .data)
-    "data",  # Executor emitted data during execution (use .executor_id, .data)
+    "output",  # Executor yielded final terminal output (use .executor_id, .data)
+    "intermediate",  # Executor emitted intermediate (non-terminal) output (use .executor_id, .data)
+    "data",  # DEPRECATED — legacy alias for intermediate emissions; use type='intermediate' instead.
     # Request events (human-in-the-loop)
     "request_info",  # Executor requests external info (use .request_id, .source_executor_id)
     # Diagnostic events (warnings/errors from user code)
@@ -128,6 +130,24 @@ WorkflowEventType = Literal[
 ]
 
 
+# Framework-managed event types — workflow lifecycle, diagnostics, and executor bookkeeping
+# — that carry no user-facing payload and are not forwarded through the
+# ``workflow.as_agent()`` boundary. Internal to the ``_workflows`` package.
+_LIFECYCLE_EVENT_TYPES: frozenset[str] = frozenset({
+    "started",
+    "status",
+    "failed",
+    "warning",
+    "error",
+    "superstep_started",
+    "superstep_completed",
+    "executor_invoked",
+    "executor_completed",
+    "executor_failed",
+    "executor_bypassed",
+})
+
+
 class WorkflowEvent(Generic[DataT]):
     """Unified event for all workflow emissions.
 
@@ -141,8 +161,8 @@ class WorkflowEvent(Generic[DataT]):
     - `WorkflowEvent.failed(details)` - workflow terminated with error
     - `WorkflowEvent.warning(message)` - warning from user code
     - `WorkflowEvent.error(exception)` - error from user code
-    - `WorkflowEvent.output(executor_id, data)` - executor yielded final output
-    - `WorkflowEvent.data(executor_id, data)` - executor emitted data (e.g., AgentResponse)
+    - `WorkflowEvent.output(executor_id, data)` - executor yielded final terminal output
+    - `WorkflowEvent.intermediate(executor_id, data)` - executor emitted intermediate (non-terminal) data
     - `WorkflowEvent.request_info(...)` - executor requests external info
     - `WorkflowEvent.superstep_started(iteration)` - superstep began
     - `WorkflowEvent.superstep_completed(iteration)` - superstep ended
@@ -265,16 +285,33 @@ class WorkflowEvent(Generic[DataT]):
 
     @classmethod
     def output(cls, executor_id: str, data: DataT) -> WorkflowEvent[DataT]:
-        """Create an 'output' event when an executor yields final output."""
+        """Create an 'output' event when an executor yields final terminal output."""
         return cls("output", executor_id=executor_id, data=data)
 
     @classmethod
-    def emit(cls, executor_id: str, data: DataT) -> WorkflowEvent[DataT]:
-        """Create a 'data' event when an executor emits data during execution.
+    def intermediate(cls, executor_id: str, data: DataT) -> WorkflowEvent[DataT]:
+        """Create an 'intermediate' event for a non-terminal emission.
 
-        This is the primary method for executors to emit typed data
-        (e.g., AgentResponse, AgentResponseUpdate, custom data).
+        The runner labels yields automatically based on the workflow's ``output_executors``;
+        this factory exists for cases that need to construct events directly.
         """
+        return cls("intermediate", executor_id=executor_id, data=data)
+
+    @classmethod
+    def emit(cls, executor_id: str, data: DataT) -> WorkflowEvent[DataT]:
+        """Create a 'data' event (deprecated alias for intermediate emissions).
+
+        .. deprecated::
+            Use :meth:`WorkflowEvent.intermediate` instead. Will be removed in a future
+            major release along with the ``type='data'`` event variant.
+        """
+        warnings.warn(
+            "WorkflowEvent.emit() / type='data' are deprecated; use WorkflowEvent.intermediate() "
+            "(or ctx.yield_output() from a non-designated executor). Will be removed in a future "
+            "major release.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
         return cls("data", executor_id=executor_id, data=data)
 
     @classmethod

python/packages/core/agent_framework/_workflows/_runner_context.py

@@ -7,7 +7,10 @@ import logging
 from copy import copy
 from dataclasses import dataclass
 from enum import Enum
-from typing import Any, Protocol, TypeVar, runtime_checkable
+from typing import TYPE_CHECKING, Any, Protocol, TypeVar, runtime_checkable
+
+if TYPE_CHECKING:
+    from ._workflow import Workflow
 
 from ._checkpoint import CheckpointID, CheckpointStorage, WorkflowCheckpoint
 from ._const import INTERNAL_SOURCE_ID
@@ -192,6 +195,19 @@ class RunnerContext(Protocol):
         """
         ...
 
+    def should_label_as_intermediate(self, executor_id: str) -> bool:
+        """Whether yields from ``executor_id`` should be labeled type='intermediate'.
+
+        Returns True only when the workflow was built with explicit ``output_executors``
+        AND ``executor_id`` is not in that designated set. In legacy mode (no explicit
+        ``output_executors``), always returns False — every yield is type='output'.
+
+        RunnerContext subclasses that don't support intermediate labeling may inherit the
+        Protocol's no-op default body, which returns ``None`` (falsy) — yielding legacy
+        behavior automatically.
+        """
+        ...
+
     async def create_checkpoint(
         self,
         workflow_name: str,
@@ -287,6 +303,12 @@ class InProcRunnerContext:
         # Streaming flag - set by workflow's run(..., stream=True) vs run(..., stream=False)
         self._streaming: bool = False
 
+        # Back-reference to the Workflow this context serves; assigned once by
+        # WorkflowBuilder.build() after both objects exist (chicken-and-egg: the workflow
+        # needs the context for its Runner). The runner consults the Workflow as the
+        # single source of truth for output designation rather than caching a copy.
+        self._workflow: Workflow | None = None
+
     # region Messaging and Events
     async def send_message(self, message: WorkflowMessage) -> None:
         self._messages.setdefault(message.source_id, [])
@@ -431,6 +453,11 @@ class InProcRunnerContext:
         """
         return self._streaming
 
+    def should_label_as_intermediate(self, executor_id: str) -> bool:
+        if self._workflow is None or self._workflow._output_executors is None:
+            return False  # not yet bound, or legacy mode
+        return executor_id not in self._workflow._output_executors
+
     async def add_request_info_event(self, event: WorkflowEvent[Any]) -> None:
         """Add a request_info event to the context and track it for correlation.
 

python/packages/core/agent_framework/_workflows/_workflow.py

@@ -198,8 +198,10 @@ class Workflow(DictConvertible):
                 better observability and management.
             description: Optional description of what the workflow does. If the workflow is built using
                 WorkflowBuilder, this will be the description of the builder.
-            output_executors: Optional list of executor IDs whose outputs will be considered workflow outputs.
-                              If None or empty, all executor outputs are treated as workflow outputs.
+            output_executors: List of executor IDs designated as terminal outputs, or
+                ``None`` for legacy mode (every yield is ``type='output'``). Any list
+                (including ``[]``) opts into strict mode where only designated yields are
+                ``type='output'``; see ``WorkflowBuilder`` for the canonical contract.
         """
         self.edge_groups = list(edge_groups)
         self.executors = dict(executors)
@@ -215,9 +217,10 @@ class Workflow(DictConvertible):
         self.graph_signature = self._compute_graph_signature()
         self.graph_signature_hash = self._hash_graph_signature(self.graph_signature)
 
-        # Output events (WorkflowEvent with type='output') from these executors are treated as workflow outputs.
-        # If None or empty, all executor outputs are considered workflow outputs.
-        self._output_executors = list(output_executors) if output_executors else list(self.executors.keys())
+        # Stored as the original nullable shape so legacy (None) and strict-no-terminals ([])
+        # are distinguishable; consumed by `_should_yield_output_event` and the runner's
+        # labeling check.
+        self._output_executors: list[str] | None = list(output_executors) if output_executors is not None else None
 
         # Store non-serializable runtime objects as private attributes
         self._runner_context = runner_context
@@ -289,7 +292,13 @@ class Workflow(DictConvertible):
         return self.executors[self.start_executor_id]
 
     def get_output_executors(self) -> list[Executor]:
-        """Get the list of output executors in the workflow."""
+        """Get the list of output executors in the workflow.
+
+        In legacy mode (no explicit ``output_executors``), returns every executor in the
+        workflow. In strict mode, returns only the designated output executors.
+        """
+        if self._output_executors is None:
+            return list(self.executors.values())
         return [self.executors[executor_id] for executor_id in self._output_executors]
 
     def get_executors_list(self) -> list[Executor]:
@@ -834,11 +843,10 @@ class Workflow(DictConvertible):
         Returns:
             True if the event should be yielded as a workflow output, False otherwise.
         """
-        # If no specific output executors are defined, yield all outputs
-        if not self._output_executors:
+        # Legacy mode: every yield is treated as a workflow output.
+        if self._output_executors is None:
             return True
-
-        # Check if the event's source executor is in the list of output executors
+        # Strict mode: only yields from designated executors are workflow outputs.
         return event.executor_id in self._output_executors
 
     # Graph signature helpers

python/packages/core/agent_framework/_workflows/_workflow_builder.py

@@ -3,6 +3,7 @@
 import logging
 import sys
 import uuid
+import warnings
 from collections.abc import Callable, Sequence
 from typing import Any
 
@@ -98,8 +99,15 @@ class WorkflowBuilder:
             start_executor: The starting executor for the workflow. Can be an Executor instance
                 or SupportsAgentRun instance.
             checkpoint_storage: Optional checkpoint storage for enabling workflow state persistence.
-            output_executors: Optional list of executors whose outputs should be collected.
-                If not provided, outputs from all executors are collected.
+            output_executors: Designates which executors emit terminal output (``type='output'``
+                workflow events). Three-state contract:
+
+                - **Unset (default):** legacy mode. Every ``yield_output`` produces ``type='output'``.
+                  A ``DeprecationWarning`` is raised at ``build()`` recommending explicit designation.
+                - **``[]`` (explicit empty list):** strict mode with no terminals. Every ``yield_output``
+                  produces ``type='intermediate'``. ``WorkflowRunResult.get_outputs()`` returns ``[]``.
+                - **``[X, ...]`` (explicit list):** strict mode. Yields from designated executors
+                  produce ``type='output'``; all other yields produce ``type='intermediate'``.
         """
         self._edge_groups: list[EdgeGroup] = []
         self._executors: dict[str, Executor] = {}
@@ -113,8 +121,12 @@ class WorkflowBuilder:
         # being created for the same agent.
         self._agent_wrappers: dict[str, Executor] = {}
 
-        # Output executors filter; if set, only outputs from these executors are yielded
-        self._output_executors: list[Executor | SupportsAgentRun] = output_executors if output_executors else []
+        # Output executors filter. ``None`` means legacy mode (every yield_output produces
+        # type='output'). Any list (including ``[]``) opts into strict mode — designated
+        # executors emit type='output', all other executors emit type='intermediate'.
+        self._output_executors: list[Executor | SupportsAgentRun] | None = (
+            list(output_executors) if output_executors is not None else None
+        )
 
         # Set the start executor
         self._set_start_executor(start_executor)
@@ -637,19 +649,39 @@ class WorkflowBuilder:
                         "Starting executor must be set via the start_executor constructor parameter before building."
                     )
 
+                if self._output_executors is None:
+                    warnings.warn(
+                        "WorkflowBuilder built without explicit output_executors; every yield_output "
+                        "produces type='output' (legacy default). Pass output_executors=[...] to opt "
+                        "into the strict contract — explicit designation will be required in a future "
+                        "major release.",
+                        DeprecationWarning,
+                        stacklevel=2,
+                    )
+
                 start_executor = self._start_executor
                 executors = self._executors
                 edge_groups = self._edge_groups
-                output_executors = [ex.id for ex in self._output_executors if isinstance(ex, Executor)] + [
-                    resolve_agent_id(agent) for agent in self._output_executors if isinstance(agent, SupportsAgentRun)
-                ]
+                # Resolve designated executor ids when output_executors was passed explicitly.
+                # Legacy mode (None) and strict mode (any list) are distinguished here once;
+                # both the runner context and the Workflow receive a single shape.
+                output_executors_for_workflow: list[str] | None = (
+                    [ex.id for ex in self._output_executors if isinstance(ex, Executor)]
+                    + [
+                        resolve_agent_id(agent)
+                        for agent in self._output_executors
+                        if isinstance(agent, SupportsAgentRun)
+                    ]
+                    if self._output_executors is not None
+                    else None
+                )
 
                 # Perform validation before creating the workflow
                 validate_workflow_graph(
                     edge_groups,
                     executors,
                     start_executor,
-                    output_executors,
+                    output_executors_for_workflow or [],
                 )
 
                 # Add validation completed event
@@ -666,8 +698,13 @@ class WorkflowBuilder:
                     self._name,
                     description=self._description,
                     max_iterations=self._max_iterations,
-                    output_executors=output_executors,
+                    output_executors=output_executors_for_workflow,
                 )
+                # Bind the runner context to the workflow so the runner can consult it as
+                # the single source of truth for output-event labeling. Both objects must
+                # exist first (the workflow needs the context for its Runner), so the
+                # back-reference is assigned here as the last step of build().
+                context._workflow = workflow
                 build_attributes: dict[str, Any] = {
                     OtelAttr.WORKFLOW_BUILDER_NAME: self._name,
                     OtelAttr.WORKFLOW_ID: workflow.id,

python/packages/core/agent_framework/_workflows/_workflow_context.py

@@ -337,7 +337,18 @@ class WorkflowContext(Generic[OutT, W_OutT]):
             await self._runner_context.send_message(msg)
 
     async def yield_output(self, output: W_OutT) -> None:
-        """Set the output of the workflow.
+        """Yield an output from this executor.
+
+        The framework labels the resulting workflow event based on whether this executor
+        is designated as an output executor (see ``WorkflowBuilder.output_executors``):
+
+        - **Strict mode** (``output_executors`` was passed explicitly to ``WorkflowBuilder``):
+
+          - If this executor is designated → ``WorkflowEvent`` with ``type='output'``.
+          - If not designated → ``WorkflowEvent`` with ``type='intermediate'``.
+
+        - **Legacy mode** (``output_executors`` unset; deprecated): every yield produces
+          ``type='output'`` regardless of executor.
 
         Args:
             output: The output to yield. This must conform to the workflow output type(s)
@@ -348,7 +359,10 @@ class WorkflowContext(Generic[OutT, W_OutT]):
         self._yielded_outputs.append(copy.deepcopy(output))
 
         with _framework_event_origin():
-            event = WorkflowEvent.output(self._executor_id, output)
+            if self._runner_context.should_label_as_intermediate(self._executor_id):
+                event = WorkflowEvent.intermediate(self._executor_id, output)
+            else:
+                event = WorkflowEvent.output(self._executor_id, output)
         await self._runner_context.add_event(event)
 
     async def add_event(self, event: WorkflowEvent[Any]) -> None:

python/packages/core/tests/workflow/test_agent_run_event_typing.py

@@ -9,7 +9,7 @@ from agent_framework._workflows._events import WorkflowEvent
 def test_workflow_event_with_agent_response_data_type() -> None:
     """Verify WorkflowEvent[AgentResponse].data is typed as AgentResponse."""
     response = AgentResponse(messages=[Message(role="assistant", contents=["Hello"])])
-    event: WorkflowEvent[AgentResponse] = WorkflowEvent.emit(executor_id="test", data=response)
+    event: WorkflowEvent[AgentResponse] = WorkflowEvent.intermediate(executor_id="test", data=response)
 
     # This assignment should pass type checking without a cast
     data: AgentResponse = event.data
@@ -20,7 +20,7 @@ def test_workflow_event_with_agent_response_data_type() -> None:
 def test_workflow_event_with_agent_response_update_data_type() -> None:
     """Verify WorkflowEvent[AgentResponseUpdate].data is typed as AgentResponseUpdate."""
     update = AgentResponseUpdate()
-    event: WorkflowEvent[AgentResponseUpdate] = WorkflowEvent.emit(executor_id="test", data=update)
+    event: WorkflowEvent[AgentResponseUpdate] = WorkflowEvent.intermediate(executor_id="test", data=update)
 
     # This assignment should pass type checking without a cast
     data: AgentResponseUpdate = event.data
@@ -30,7 +30,7 @@ def test_workflow_event_with_agent_response_update_data_type() -> None:
 def test_workflow_event_repr() -> None:
     """Verify WorkflowEvent.__repr__ uses consistent format."""
     response = AgentResponse(messages=[Message(role="assistant", contents=["Hello"])])
-    event: WorkflowEvent[AgentResponse] = WorkflowEvent.emit(executor_id="test", data=response)
+    event: WorkflowEvent[AgentResponse] = WorkflowEvent.intermediate(executor_id="test", data=response)
 
     repr_str = repr(event)
     assert "WorkflowEvent" in repr_str

python/packages/core/tests/workflow/test_functional_workflow.py

@@ -165,13 +165,13 @@ class TestEventEmission:
 
         @workflow
         async def pipeline(x: int, ctx: RunContext) -> int:
-            await ctx.add_event(WorkflowEvent.emit("pipeline", "custom_data"))
+            await ctx.add_event(WorkflowEvent.intermediate("pipeline", "custom_data"))
             return x
 
         result = await pipeline.run(1)
-        data_events = [e for e in result if e.type == "data"]
-        assert len(data_events) == 1
-        assert data_events[0].data == "custom_data"
+        intermediate_events = [e for e in result if e.type == "intermediate"]
+        assert len(intermediate_events) == 1
+        assert intermediate_events[0].data == "custom_data"
 
 
 # ---------------------------------------------------------------------------

python/packages/core/tests/workflow/test_output_executors_contract.py

@@ -0,0 +1,41 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Tests for the three-state output_executors contract on WorkflowBuilder.
+
+State A: output_executors=None (unset, legacy)  -> DeprecationWarning at build
+State B: output_executors=[] (explicit, no terminals) -> strict mode
+State C: output_executors=[X, ...] (explicit list) -> strict mode
+"""
+
+from __future__ import annotations
+
+import warnings
+
+import pytest
+from typing_extensions import Never
+
+from agent_framework import (
+    Message,
+    WorkflowBuilder,
+    WorkflowContext,
+    executor,
+)
+
+
+@executor
+async def _emit_one(messages: list[Message], ctx: WorkflowContext[Never, str]) -> None:
+    await ctx.yield_output("hello")
+
+
+def test_output_executors_unset_emits_deprecation_warning() -> None:
+    """State A: WorkflowBuilder built without explicit output_executors warns."""
+    with pytest.warns(DeprecationWarning, match="output_executors"):
+        WorkflowBuilder(start_executor=_emit_one).build()
+
+
+@pytest.mark.parametrize("output_executors", [[], [_emit_one]], ids=["empty_list", "designated_list"])
+def test_output_executors_explicit_value_does_not_warn(output_executors) -> None:
+    """States B and C: any explicit list (including ``[]``) opts into strict mode without warning."""
+    with warnings.catch_warnings():
+        warnings.simplefilter("error", DeprecationWarning)
+        WorkflowBuilder(start_executor=_emit_one, output_executors=output_executors).build()

python/packages/core/tests/workflow/test_strict_mode_event_labeling.py

@@ -0,0 +1,112 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Tests for the runner's strict-mode event labeling.
+
+Strict mode = WorkflowBuilder built with explicit output_executors=[...].
+- Yields from designated executors -> type='output'.
+- Yields from non-designated executors -> type='intermediate'.
+
+Legacy mode (output_executors unset) preserves today's behavior:
+every yield -> type='output'.
+"""
+
+from __future__ import annotations
+
+import warnings
+from typing import Any
+
+import pytest
+from typing_extensions import Never
+
+from agent_framework import (
+    Message,
+    WorkflowBuilder,
+    WorkflowContext,
+    executor,
+)
+
+
+@executor
+async def _start(messages: list[Message], ctx: WorkflowContext[str, str]) -> None:
+    await ctx.yield_output("from-start")
+    await ctx.send_message("downstream")
+
+
+@executor
+async def _downstream(message: str, ctx: WorkflowContext[Never, str]) -> None:
+    await ctx.yield_output("from-downstream")
+
+
+def _input_msg() -> list[Message]:
+    return [Message(role="user", contents=["hi"])]
+
+
+@pytest.mark.asyncio
+async def test_strict_mode_designated_executor_emits_output_events() -> None:
+    """In strict mode, the designated executor's yields produce type='output' events."""
+    workflow = WorkflowBuilder(start_executor=_start, output_executors=[_start]).add_edge(_start, _downstream).build()
+    output_events: list[Any] = []
+    intermediate_events: list[Any] = []
+    async for event in workflow.run(_input_msg(), stream=True):
+        if event.type == "output":
+            output_events.append(event)
+        elif event.type == "intermediate":
+            intermediate_events.append(event)
+
+    assert any(ev.data == "from-start" for ev in output_events), "designated executor's yield is type='output'"
+    assert any(ev.data == "from-downstream" for ev in intermediate_events), (
+        "non-designated executor's yield is relabeled to type='intermediate'"
+    )
+
+
+@pytest.mark.asyncio
+async def test_strict_mode_empty_list_means_no_terminals() -> None:
+    """Strict mode with output_executors=[] produces zero type='output' events; everything is intermediate."""
+    workflow = WorkflowBuilder(start_executor=_start, output_executors=[]).add_edge(_start, _downstream).build()
+    output_events: list[Any] = []
+    intermediate_events: list[Any] = []
+    async for event in workflow.run(_input_msg(), stream=True):
+        if event.type == "output":
+            output_events.append(event)
+        elif event.type == "intermediate":
+            intermediate_events.append(event)
+
+    assert len(output_events) == 0
+    assert {ev.data for ev in intermediate_events} == {"from-start", "from-downstream"}
+
+
+@pytest.mark.asyncio
+async def test_legacy_mode_unset_keeps_all_yields_as_output() -> None:
+    """Legacy mode (output_executors unset) preserves today's behavior — all yields are type='output'."""
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", DeprecationWarning)
+        workflow = WorkflowBuilder(start_executor=_start).add_edge(_start, _downstream).build()
+    output_events: list[Any] = []
+    intermediate_events: list[Any] = []
+    async for event in workflow.run(_input_msg(), stream=True):
+        if event.type == "output":
+            output_events.append(event)
+        elif event.type == "intermediate":
+            intermediate_events.append(event)
+
+    assert {ev.data for ev in output_events} == {"from-start", "from-downstream"}
+    assert len(intermediate_events) == 0
+
+
+@pytest.mark.asyncio
+async def test_strict_mode_get_outputs_returns_only_designated() -> None:
+    """WorkflowRunResult.get_outputs() returns only designated outputs in strict mode."""
+    workflow = (
+        WorkflowBuilder(start_executor=_start, output_executors=[_downstream]).add_edge(_start, _downstream).build()
+    )
+    result = await workflow.run(_input_msg())
+    outputs = result.get_outputs()
+    assert outputs == ["from-downstream"]
+
+
+@pytest.mark.asyncio
+async def test_strict_mode_get_outputs_empty_with_no_terminals() -> None:
+    """output_executors=[] yields no terminal outputs."""
+    workflow = WorkflowBuilder(start_executor=_start, output_executors=[]).add_edge(_start, _downstream).build()
+    result = await workflow.run(_input_msg())
+    assert result.get_outputs() == []

python/packages/core/tests/workflow/test_validation.py

@@ -609,15 +609,20 @@ def test_output_validation_fails_for_executor_without_output_types():
 
 
 def test_output_validation_empty_list_passes():
-    """Test that output validation passes with an empty output executors list."""
+    """Test that output validation passes with an explicit empty output executors list.
+
+    Under the strict-output contract, ``output_executors=[]`` means "no terminals" —
+    no executor's yield is type='output'. This is distinct from the legacy default
+    (``output_executors=None``) which treats every executor as an output.
+    """
     executor1 = OutputExecutor(id="executor1")
     executor2 = OutputExecutor(id="executor2")
 
     workflow = WorkflowBuilder(start_executor=executor1, output_executors=[]).add_edge(executor1, executor2).build()
 
     assert workflow is not None
-    # All executors are outputs
-    assert workflow._output_executors == ["executor1", "executor2"]  # type: ignore
+    # Explicit empty list = strict mode with zero designated outputs.
+    assert workflow._output_executors == []  # type: ignore[attr-defined]
 
 
 def test_output_validation_with_direct_validate_workflow_graph():

python/packages/core/tests/workflow/test_workflow_agent_intermediate.py

@@ -0,0 +1,158 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Tests for WorkflowAgent translation of intermediate events to text_reasoning content.
+
+Covers:
+- type='intermediate' surfaces as AgentResponseUpdate with text_reasoning content
+- type='data' (legacy via WorkflowEvent.emit) surfaces the same way (fixes F7)
+- update.text returns terminal-only by virtue of the existing content-type filter
+- Message.additional_properties survives the intermediate translation path
+- Terminal yields keep using regular text content (backward compat)
+"""
+
+from __future__ import annotations
+
+import warnings
+
+import pytest
+from typing_extensions import Never
+
+from agent_framework import (
+    AgentResponse,
+    AgentResponseUpdate,
+    Content,
+    Message,
+    WorkflowBuilder,
+    WorkflowContext,
+    WorkflowEvent,
+    executor,
+)
+
+
+@pytest.mark.asyncio
+async def test_workflow_agent_forwards_intermediate_events_as_text_reasoning() -> None:
+    """An intermediate yield from a non-designated executor surfaces through as_agent
+    as an AgentResponseUpdate carrying text_reasoning content."""
+
+    @executor
+    async def emit(messages: list[Message], ctx: WorkflowContext[str, str]) -> None:
+        await ctx.yield_output("intermediate progress")
+        await ctx.send_message("downstream")
+
+    @executor
+    async def terminal(message: str, ctx: WorkflowContext[Never, str]) -> None:
+        await ctx.yield_output("FINAL")
+
+    workflow = WorkflowBuilder(start_executor=emit, output_executors=[terminal]).add_edge(emit, terminal).build()
+    agent = workflow.as_agent("test")
+
+    updates: list[AgentResponseUpdate] = []
+    async for update in agent.run("hi", stream=True):
+        updates.append(update)
+
+    # Find the intermediate progress: it has text_reasoning content.
+    intermediate_updates = [u for u in updates if any(c.type == "text_reasoning" for c in u.contents)]
+    terminal_updates = [u for u in updates if any(c.type == "text" for c in u.contents)]
+
+    intermediate_text = " ".join(c.text for u in intermediate_updates for c in u.contents if c.type == "text_reasoning")
+    terminal_text = " ".join(u.text for u in terminal_updates)
+
+    assert "intermediate progress" in intermediate_text
+    assert "FINAL" in terminal_text
+
+
+@pytest.mark.asyncio
+async def test_workflow_agent_text_accessor_returns_terminal_only() -> None:
+    """update.text excludes text_reasoning content automatically. The non-streaming
+    AgentResponse.text returns only terminal text — intermediate progress is invisible
+    to existing callers using .text."""
+
+    @executor
+    async def emit(messages: list[Message], ctx: WorkflowContext[str, str]) -> None:
+        await ctx.yield_output("invisible-progress")
+        await ctx.send_message("forward")
+
+    @executor
+    async def terminal(message: str, ctx: WorkflowContext[Never, str]) -> None:
+        await ctx.yield_output("the-answer")
+
+    workflow = WorkflowBuilder(start_executor=emit, output_executors=[terminal]).add_edge(emit, terminal).build()
+    agent = workflow.as_agent("test")
+
+    response = await agent.run("hi")
+    assert isinstance(response, AgentResponse)
+    # .text filters to content.type == "text" — intermediate text_reasoning is excluded.
+    assert response.text == "the-answer"
+
+
+@pytest.mark.asyncio
+async def test_workflow_agent_legacy_data_event_emit_factory_still_forwarded() -> None:
+    """Even the deprecated WorkflowEvent.emit() / type='data' path is forwarded as
+    text_reasoning content (was previously dropped — F7 fix)."""
+
+    @executor
+    async def emit_legacy(messages: list[Message], ctx: WorkflowContext[Never, str]) -> None:
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore", DeprecationWarning)
+            await ctx.add_event(WorkflowEvent.emit("emit_legacy", "legacy-payload"))
+        await ctx.yield_output("DONE")
+
+    workflow = WorkflowBuilder(start_executor=emit_legacy, output_executors=[emit_legacy]).build()
+    agent = workflow.as_agent("test")
+
+    updates: list[AgentResponseUpdate] = []
+    async for update in agent.run("hi", stream=True):
+        updates.append(update)
+
+    reasoning_text = " ".join(c.text for u in updates for c in u.contents if c.type == "text_reasoning")
+    assert "legacy-payload" in reasoning_text
+
+
+@pytest.mark.asyncio
+async def test_workflow_agent_intermediate_message_preserves_additional_properties() -> None:
+    """Message.additional_properties survives the intermediate translation path.
+
+    Regression test for the omitted field in _mark_msg — without forwarding
+    additional_properties, producer-attached metadata (tracking_id, conversation_id, etc.)
+    silently disappears for messages flowing through non-designated executors.
+    """
+
+    @executor
+    async def emit(messages: list[Message], ctx: WorkflowContext[str, AgentResponse]) -> None:
+        msg = Message(
+            role="assistant",
+            contents=[Content.from_text(text="hi")],
+            additional_properties={"tracking_id": "abc-123"},
+        )
+        await ctx.yield_output(AgentResponse(messages=[msg]))
+        await ctx.send_message("forward")
+
+    @executor
+    async def terminal(message: str, ctx: WorkflowContext[Never, str]) -> None:
+        await ctx.yield_output("done")
+
+    workflow = WorkflowBuilder(start_executor=emit, output_executors=[terminal]).add_edge(emit, terminal).build()
+    agent = workflow.as_agent("test")
+
+    response = await agent.run("hi")
+    intermediate_msgs = [m for m in response.messages if any(c.type == "text_reasoning" for c in m.contents)]
+    assert intermediate_msgs, "expected at least one intermediate message in the response"
+    assert intermediate_msgs[0].additional_properties.get("tracking_id") == "abc-123"
+
+
+@pytest.mark.asyncio
+async def test_workflow_agent_terminal_text_stays_text_not_reasoning() -> None:
+    """Backward compat — a designated executor's text yield surfaces as Content.text,
+    not text_reasoning. Existing consumers reading .text on the response work unchanged."""
+
+    @executor
+    async def only(messages: list[Message], ctx: WorkflowContext[Never, str]) -> None:
+        await ctx.yield_output("the-answer")
+
+    workflow = WorkflowBuilder(start_executor=only, output_executors=[only]).build()
+    agent = workflow.as_agent("test")
+
+    response = await agent.run("hi")
+    assert response.text == "the-answer"
+    # No text_reasoning content because everything from `only` is terminal.
+    assert all(c.type != "text_reasoning" for m in response.messages for c in m.contents)

python/packages/core/tests/workflow/test_workflow_event_factories.py

@@ -0,0 +1,39 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Tests for WorkflowEvent factory methods, including the new intermediate factory
+and the deprecation of WorkflowEvent.emit() / type='data'."""
+
+from __future__ import annotations
+
+import warnings
+
+import pytest
+
+from agent_framework import AgentResponse, Message
+from agent_framework._workflows._events import WorkflowEvent
+
+
+def test_workflow_event_intermediate_factory_creates_intermediate_event() -> None:
+    """WorkflowEvent.intermediate(executor_id, data) creates a type='intermediate' event."""
+    response = AgentResponse(messages=[Message(role="assistant", contents=["Hello"])])
+    event: WorkflowEvent[AgentResponse] = WorkflowEvent.intermediate(executor_id="test", data=response)
+
+    assert event.type == "intermediate"
+    assert event.executor_id == "test"
+    assert event.data is response
+
+
+def test_workflow_event_emit_emits_deprecation_warning() -> None:
+    """Calling WorkflowEvent.emit() raises a DeprecationWarning recommending the new path."""
+    response = AgentResponse(messages=[Message(role="assistant", contents=["x"])])
+    with pytest.warns(DeprecationWarning, match="intermediate"):
+        WorkflowEvent.emit(executor_id="t", data=response)
+
+
+def test_workflow_event_emit_still_returns_data_event() -> None:
+    """During the deprecation window, emit() still produces a type='data' event."""
+    response = AgentResponse(messages=[Message(role="assistant", contents=["x"])])
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", DeprecationWarning)
+        event = WorkflowEvent.emit(executor_id="t", data=response)
+    assert event.type == "data"

python/packages/devui/agent_framework_devui/_mapper.py

@@ -200,10 +200,12 @@ class MessageMapper:
         try:
             from agent_framework import AgentResponse, AgentResponseUpdate, WorkflowEvent
 
-            # Handle WorkflowEvent with type='output' or 'data' wrapping AgentResponseUpdate
-            # This must be checked BEFORE generic WorkflowEvent check
-            # Note: AgentExecutor uses type='output' for streaming updates
-            if isinstance(raw_event, WorkflowEvent) and raw_event.type in ("output", "data"):
+            # Handle WorkflowEvent with type='output', 'intermediate', or 'data' wrapping
+            # AgentResponseUpdate. This must be checked BEFORE generic WorkflowEvent check.
+            # Note: AgentExecutor uses type='output' for streaming updates from designated
+            # executors and type='intermediate' from non-designated executors. type='data'
+            # is the deprecated legacy variant retained for backward compat.
+            if isinstance(raw_event, WorkflowEvent) and raw_event.type in ("output", "intermediate", "data"):
                 event_data = getattr(cast(Any, raw_event), "data", None)
                 if isinstance(event_data, AgentResponseUpdate):
                     # Preserve executor_id in context for proper output routing

python/packages/devui/tests/devui/test_mapper.py

@@ -429,7 +429,7 @@ async def test_magentic_executor_event_with_agent_delta_metadata(
     """Test that WorkflowEvent[AgentResponseUpdate] with magentic_event_type='agent_delta' is handled correctly.
 
     This tests the ACTUAL event format Magentic emits - not a fake MagenticAgentDeltaEvent class.
-    Magentic uses WorkflowEvent.emit() with additional_properties containing magentic_event_type.
+    Magentic uses WorkflowEvent.intermediate() with additional_properties containing magentic_event_type.
     """
     from agent_framework._types import AgentResponseUpdate
     from agent_framework._workflows._events import WorkflowEvent
@@ -444,7 +444,7 @@ async def test_magentic_executor_event_with_agent_delta_metadata(
             "agent_id": "writer_agent",
         },
     )
-    event = WorkflowEvent.emit(executor_id="magentic_executor", data=update)
+    event = WorkflowEvent.intermediate(executor_id="magentic_executor", data=update)
 
     events = await mapper.convert_event(event, test_request)
 
@@ -459,7 +459,7 @@ async def test_magentic_executor_event_with_agent_delta_metadata(
 async def test_magentic_orchestrator_message_event(mapper: MessageMapper, test_request: AgentFrameworkRequest) -> None:
     """Test that WorkflowEvent[AgentResponseUpdate] with magentic_event_type='orchestrator_message' is handled.
 
-    Magentic emits orchestrator planning/instruction messages using WorkflowEvent.emit()
+    Magentic emits orchestrator planning/instruction messages using WorkflowEvent.intermediate()
     with additional_properties containing magentic_event_type='orchestrator_message'.
     """
     from agent_framework._types import AgentResponseUpdate
@@ -476,7 +476,7 @@ async def test_magentic_orchestrator_message_event(mapper: MessageMapper, test_r
             "orchestrator_id": "magentic_orchestrator",
         },
     )
-    event = WorkflowEvent.emit(executor_id="magentic_orchestrator", data=update)
+    event = WorkflowEvent.intermediate(executor_id="magentic_orchestrator", data=update)
 
     events = await mapper.convert_event(event, test_request)
 
@@ -507,7 +507,7 @@ async def test_magentic_events_use_same_event_class_as_other_workflows(
         contents=[Content.from_text(text="Regular workflow response")],
         role="assistant",
     )
-    regular_event = WorkflowEvent.emit(executor_id="regular_executor", data=regular_update)
+    regular_event = WorkflowEvent.intermediate(executor_id="regular_executor", data=regular_update)
 
     # 2. Magentic workflow (with additional_properties)
     magentic_update = AgentResponseUpdate(
@@ -515,7 +515,7 @@ async def test_magentic_events_use_same_event_class_as_other_workflows(
         role="assistant",
         additional_properties={"magentic_event_type": "agent_delta"},
     )
-    magentic_event = WorkflowEvent.emit(executor_id="magentic_executor", data=magentic_update)
+    magentic_event = WorkflowEvent.intermediate(executor_id="magentic_executor", data=magentic_update)
 
     # Both should be the SAME class
     assert type(regular_event) is type(magentic_event)
@@ -594,6 +594,27 @@ async def test_workflow_output_event_with_list_data(mapper: MessageMapper, test_
     assert events[0].type == "response.output_item.added"
 
 
+async def test_workflow_intermediate_event_with_agent_response_update_dispatched(
+    mapper: MessageMapper, test_request: AgentFrameworkRequest
+) -> None:
+    """A WorkflowEvent with type='intermediate' wrapping an AgentResponseUpdate is mapped
+    just like type='output' / type='data' — to OpenAI text-delta events."""
+    from agent_framework._workflows._events import WorkflowEvent
+
+    update = AgentResponseUpdate(
+        contents=[Content.from_text(text="intermediate progress")],
+        role="assistant",
+        author_name="non-designated-agent",
+    )
+    event = WorkflowEvent.intermediate(executor_id="non_designated", data=update)
+    events = await mapper.convert_event(event, test_request)
+
+    assert len(events) >= 1
+    text_events = [e for e in events if getattr(e, "type", "") == "response.output_text.delta"]
+    assert len(text_events) >= 1
+    assert text_events[0].delta == "intermediate progress"
+
+
 # =============================================================================
 # failed event (type='failed') Tests
 # =============================================================================

python/packages/orchestrations/agent_framework_orchestrations/_concurrent.py

@@ -396,10 +396,16 @@ class ConcurrentBuilder:
         # Resolve participants and participant factories to executors
         participants: list[Executor] = self._resolve_participants()
 
+        # Default: only the aggregator is designated; participants surface as intermediate.
+        # With intermediate_outputs=True, participants are also designated so their yields
+        # surface as type='output' — preserving the legacy "see every participant" contract.
+        designated: list[Executor | SupportsAgentRun] = (
+            [aggregator, *participants] if self._intermediate_outputs else [aggregator]
+        )
         builder = WorkflowBuilder(
             start_executor=dispatcher,
             checkpoint_storage=self._checkpoint_storage,
-            output_executors=[aggregator] if not self._intermediate_outputs else None,
+            output_executors=designated,
         )
         # Fan-out for parallel execution
         builder.add_fan_out_edges(dispatcher, participants)

python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py

@@ -1001,11 +1001,17 @@ class GroupChatBuilder:
         participants: list[Executor] = self._resolve_participants()
         orchestrator: Executor = self._resolve_orchestrator(participants)
 
-        # Build workflow graph
+        # Default: only the orchestrator is designated; participants surface as intermediate.
+        # With intermediate_outputs=True, participants are also designated so their yields
+        # surface as type='output' — preserving the legacy contract.
+        # `group_chat` orchestrator-progress events keep their dedicated event type.
+        designated: list[Executor | SupportsAgentRun] = (
+            [orchestrator, *participants] if self._intermediate_outputs else [orchestrator]
+        )
         workflow_builder = WorkflowBuilder(
             start_executor=orchestrator,
             checkpoint_storage=self._checkpoint_storage,
-            output_executors=[orchestrator] if not self._intermediate_outputs else None,
+            output_executors=designated,
         )
         for participant in participants:
             # Orchestrator and participant bi-directional edges

python/packages/orchestrations/agent_framework_orchestrations/_handoff.py

@@ -955,11 +955,15 @@ class HandoffBuilder:
         if self._start_id is None:
             raise ValueError("Must call with_start_agent(...) before building the workflow.")
         start_executor = executors[self._resolve_to_id(resolved_agents[self._start_id])]
+        # Handoff has no separate terminator: every participant's reply is a primary
+        # output (termination is implicit when no handoff fires). All participants are
+        # designated outputs so each yield surfaces as type='output'.
         builder = WorkflowBuilder(
             name=self._name,
             description=self._description,
             start_executor=start_executor,
             checkpoint_storage=self._checkpoint_storage,
+            output_executors=list(executors.values()),
         )
 
         # Add the appropriate edges

python/packages/orchestrations/agent_framework_orchestrations/_magentic.py

@@ -1762,11 +1762,17 @@ class MagenticBuilder:
         participants: list[Executor] = self._resolve_participants()
         orchestrator: Executor = self._resolve_orchestrator(participants)
 
-        # Build workflow graph
+        # Default: only the manager is designated; worker yields surface as intermediate.
+        # With intermediate_outputs=True, workers are also designated so their yields
+        # surface as type='output' — preserving the legacy contract.
+        # `magentic_orchestrator` events keep their dedicated event type.
+        designated: list[Executor | SupportsAgentRun] = (
+            [orchestrator, *participants] if self._intermediate_outputs else [orchestrator]
+        )
         workflow_builder = WorkflowBuilder(
             start_executor=orchestrator,
             checkpoint_storage=self._checkpoint_storage,
-            output_executors=[orchestrator] if not self._intermediate_outputs else None,
+            output_executors=designated,
         )
         for participant in participants:
             # Orchestrator and participant bi-directional edges

python/packages/orchestrations/agent_framework_orchestrations/_orchestration_request_info.py

@@ -220,8 +220,14 @@ class AgentApprovalExecutor(WorkflowExecutor):
         request_info_cls = _TerminalAgentRequestInfoExecutor if terminal else AgentRequestInfoExecutor
         request_info_executor = request_info_cls(id="agent_request_info_executor")
 
+        # Both inner executors yield the inner workflow's terminal output (the agent
+        # during its turn; the _TerminalAgentRequestInfoExecutor after approval), so
+        # both must be designated for WorkflowExecutor.get_outputs() to surface them.
         return (
-            WorkflowBuilder(start_executor=agent_executor)
+            WorkflowBuilder(
+                start_executor=agent_executor,
+                output_executors=[agent_executor, request_info_executor],
+            )
             # Create a loop between agent executor and request info executor
             .add_edge(agent_executor, request_info_executor)
             .add_edge(request_info_executor, agent_executor)

python/packages/orchestrations/agent_framework_orchestrations/_sequential.py

@@ -234,10 +234,17 @@ class SequentialBuilder:
         # Resolve participants and participant factories to executors
         participants: list[Executor] = self._resolve_participants()
 
+        # Default: only the terminator is designated; earlier participants' yields
+        # surface as type='intermediate'. With intermediate_outputs=True, every
+        # participant is designated so all yields surface as type='output' — preserving
+        # the legacy contract for callers who opt in to seeing per-participant outputs.
+        designated: list[Executor | SupportsAgentRun] = (
+            list(participants) if self._intermediate_outputs else [participants[-1]]
+        )
         builder = WorkflowBuilder(
             start_executor=input_conv,
             checkpoint_storage=self._checkpoint_storage,
-            output_executors=[participants[-1]] if not self._intermediate_outputs else None,
+            output_executors=designated,
         )
 
         prior: Executor | SupportsAgentRun = input_conv

python/packages/orchestrations/tests/test_magentic.py

@@ -632,7 +632,8 @@ 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
+    # With intermediate_outputs=True, participants are designated as outputs alongside
+    # the manager — so their streaming chunks surface as type='output' (not intermediate).
     events: list[WorkflowEvent] = []
     async for ev in wf.run("task", stream=True):
         events.append(ev)

python/packages/orchestrations/tests/test_orchestration_intermediate_vs_terminal.py

@@ -0,0 +1,259 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Tests for orchestration intermediate vs terminal output labeling.
+
+Verifies that under the strict-output model:
+  - Sequential / Concurrent / GroupChat / Magentic designate their terminator,
+    aggregator, orchestrator, or manager as the sole output executor; per-step
+    yields from non-designated executors emit `type='intermediate'` events.
+  - Handoff designates ALL participants — every reply is `type='output'`.
+  - When wrapped via `workflow.as_agent()`, intermediate events surface as
+    `text_reasoning` content; terminal events as `text` content; existing
+    `.text` accessors return terminal-only.
+"""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterable, Awaitable
+from typing import Any, Literal, overload
+
+import pytest
+from agent_framework import (
+    AgentResponse,
+    AgentResponseUpdate,
+    AgentRunInputs,
+    AgentSession,
+    BaseAgent,
+    Content,
+    Message,
+    ResponseStream,
+)
+from agent_framework.orchestrations import ConcurrentBuilder, SequentialBuilder
+
+
+class _EchoAgent(BaseAgent):
+    """Minimal non-streaming agent that returns a single assistant message."""
+
+    @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")], author_name=self.name
+                )
+
+            return ResponseStream(_stream(), finalizer=AgentResponse.from_updates)
+
+        async def _run() -> AgentResponse:
+            return AgentResponse(messages=[Message("assistant", [f"{self.name} reply"], author_name=self.name)])
+
+        return _run()
+
+
+# ---------------------------------------------------------------------------
+# Sequential
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_sequential_default_only_terminator_is_output() -> None:
+    """Default Sequential (intermediate_outputs=False) designates only the terminator;
+    earlier participants surface as type='intermediate'."""
+    a = _EchoAgent(name="A")
+    b = _EchoAgent(name="B")
+    c = _EchoAgent(name="C")
+
+    workflow = SequentialBuilder(participants=[a, b, c]).build()
+
+    output_events: list[Any] = []
+    intermediate_events: list[Any] = []
+    async for event in workflow.run("hello", stream=True):
+        if event.type == "output":
+            output_events.append(event)
+        elif event.type == "intermediate":
+            intermediate_events.append(event)
+
+    # Only the terminator (C) emits type='output'.
+    assert len(output_events) == 1
+    assert "C" in {ev.executor_id for ev in output_events}
+
+    # A and B emit type='intermediate'.
+    intermediate_executors = {ev.executor_id for ev in intermediate_events}
+    assert "A" in intermediate_executors
+    assert "B" in intermediate_executors
+
+
+@pytest.mark.asyncio
+async def test_sequential_intermediate_outputs_true_designates_all() -> None:
+    """Sequential with intermediate_outputs=True preserves the legacy contract:
+    every participant's yield surfaces as type='output'."""
+    a = _EchoAgent(name="A")
+    b = _EchoAgent(name="B")
+    c = _EchoAgent(name="C")
+
+    workflow = SequentialBuilder(participants=[a, b, c], intermediate_outputs=True).build()
+    result = await workflow.run("hello")
+    outputs = result.get_outputs()
+    # All three participants' yields surface in get_outputs() under intermediate_outputs=True.
+    assert len(outputs) == 3
+
+
+@pytest.mark.asyncio
+async def test_sequential_get_outputs_returns_terminator_only() -> None:
+    """WorkflowRunResult.get_outputs() returns only the terminator's yield."""
+    a = _EchoAgent(name="A")
+    b = _EchoAgent(name="B")
+
+    workflow = SequentialBuilder(participants=[a, b]).build()
+    result = await workflow.run("hi")
+    outputs = result.get_outputs()
+    assert len(outputs) == 1
+
+
+# ---------------------------------------------------------------------------
+# Concurrent
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_concurrent_default_only_aggregator_is_output() -> None:
+    """Default Concurrent (intermediate_outputs=False): only the aggregator is
+    designated; participants surface as type='intermediate'."""
+    a = _EchoAgent(name="A")
+    b = _EchoAgent(name="B")
+
+    workflow = ConcurrentBuilder(participants=[a, b]).build()
+
+    output_events: list[Any] = []
+    intermediate_events: list[Any] = []
+    async for event in workflow.run("hello", stream=True):
+        if event.type == "output":
+            output_events.append(event)
+        elif event.type == "intermediate":
+            intermediate_events.append(event)
+
+    # Aggregator is the only designated executor → only it emits type='output'.
+    assert len(output_events) == 1
+
+    # Both participants emit type='intermediate'.
+    intermediate_authors = {ev.executor_id for ev in intermediate_events}
+    assert "A" in intermediate_authors
+    assert "B" in intermediate_authors
+
+
+@pytest.mark.asyncio
+async def test_concurrent_intermediate_outputs_true_designates_all() -> None:
+    """Concurrent with intermediate_outputs=True designates participants alongside the
+    aggregator — every participant's yield surfaces as type='output'."""
+    a = _EchoAgent(name="A")
+    b = _EchoAgent(name="B")
+
+    workflow = ConcurrentBuilder(participants=[a, b], intermediate_outputs=True).build()
+    result = await workflow.run("hello")
+    outputs = result.get_outputs()
+    # Two participants + aggregator → three terminal outputs in get_outputs().
+    assert len(outputs) == 3
+
+
+# ---------------------------------------------------------------------------
+# Sequential wrapped as_agent — text_reasoning mapping
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_sequential_default_as_agent_intermediates_are_text_reasoning() -> None:
+    """Default Sequential wrapped as_agent: per-step participant replies become
+    text_reasoning content; the terminator's reply becomes text content.
+    """
+    a = _EchoAgent(name="A")
+    b = _EchoAgent(name="B")
+    c = _EchoAgent(name="C")
+
+    workflow = SequentialBuilder(participants=[a, b, c]).build()
+    agent = workflow.as_agent("seq")
+
+    response = await agent.run("hi")
+
+    # .text returns terminal content only — only C's reply.
+    assert response.text == "C reply"
+
+    text_contents = [c for m in response.messages for c in m.contents if c.type == "text"]
+    reasoning_contents = [c for m in response.messages for c in m.contents if c.type == "text_reasoning"]
+
+    assert any("C reply" in c.text for c in text_contents)
+    assert any("A reply" in c.text for c in reasoning_contents)
+    assert any("B reply" in c.text for c in reasoning_contents)
+
+
+@pytest.mark.asyncio
+async def test_sequential_as_agent_intermediate_outputs_true_all_text() -> None:
+    """Sequential with intermediate_outputs=True wrapped as_agent: every participant's
+    reply is now designated terminal, so each surfaces as text content (not reasoning).
+    Existing callers reading .text get all participants' replies concatenated."""
+    a = _EchoAgent(name="A")
+    b = _EchoAgent(name="B")
+    c = _EchoAgent(name="C")
+
+    workflow = SequentialBuilder(participants=[a, b, c], intermediate_outputs=True).build()
+    agent = workflow.as_agent("seq")
+
+    response = await agent.run("hi")
+    text_contents = [c for m in response.messages for c in m.contents if c.type == "text"]
+    text = " ".join(c.text for c in text_contents)
+    assert "A reply" in text
+    assert "B reply" in text
+    assert "C reply" in text
+
+
+# ---------------------------------------------------------------------------
+# Concurrent wrapped as_agent
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_concurrent_default_as_agent_participants_are_text_reasoning() -> None:
+    """Default Concurrent wrapped as_agent: participant replies are text_reasoning;
+    aggregator's yield is text content."""
+    a = _EchoAgent(name="A")
+    b = _EchoAgent(name="B")
+
+    workflow = ConcurrentBuilder(participants=[a, b]).build()
+    agent = workflow.as_agent("concurrent")
+
+    response = await agent.run("hi")
+
+    text_contents = [c for m in response.messages for c in m.contents if c.type == "text"]
+    reasoning_contents = [c for m in response.messages for c in m.contents if c.type == "text_reasoning"]
+
+    # A's and B's replies are intermediate (text_reasoning).
+    assert any("A reply" in c.text for c in reasoning_contents)
+    assert any("B reply" in c.text for c in reasoning_contents)
+
+    # The aggregator's default-yielded AgentResponse passes through as text content.
+    assert text_contents, "expected at least one terminal text content from the aggregator"

python/samples/03-workflows/README.md

@@ -89,6 +89,7 @@ Write workflows as plain Python async functions — no graph concepts, no execut
 | Multi-Selection Edge Group | [control-flow/multi_selection_edge_group.py](./control-flow/multi_selection_edge_group.py) | Select one or many targets dynamically (subset fan-out) |
 | Simple Loop                | [control-flow/simple_loop.py](./control-flow/simple_loop.py)                               | Feedback loop where an agent judges ABOVE/BELOW/MATCHED |
 | Workflow Cancellation      | [control-flow/workflow_cancellation.py](./control-flow/workflow_cancellation.py)           | Cancel a running workflow using asyncio tasks           |
+| Intermediate vs Terminal Outputs | [control-flow/intermediate_vs_terminal_outputs.py](./control-flow/intermediate_vs_terminal_outputs.py) | Designate output executors so non-designated yields surface as `type='intermediate'` events (and `text_reasoning` content via `as_agent`) |
 
 ### human-in-the-loop
 

python/samples/03-workflows/control-flow/intermediate_vs_terminal_outputs.py

@@ -0,0 +1,118 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+
+from agent_framework import (
+    Message,
+    WorkflowBuilder,
+    WorkflowContext,
+    executor,
+)
+from typing_extensions import Never
+
+"""
+Sample: Intermediate vs terminal output labeling
+
+What this sample shows
+- How ``WorkflowBuilder(output_executors=[...])`` designates which executors emit
+  the workflow's terminal output.
+- How yields from non-designated executors automatically surface as
+  ``type='intermediate'`` events, while yields from designated executors surface
+  as ``type='output'``.
+- How the same workflow wrapped via ``workflow.as_agent()`` translates intermediate
+  events to ``text_reasoning`` content so existing ``.text`` accessors keep
+  returning terminal-only output.
+
+The contract on ``output_executors``:
+- ``None`` (default, legacy): every ``yield_output`` produces ``type='output'``.
+  Emits a ``DeprecationWarning`` at build() time recommending explicit designation.
+- ``[]`` (explicit empty list): strict mode, no terminals. Every yield is
+  ``type='intermediate'``; ``WorkflowRunResult.get_outputs()`` returns ``[]``.
+- ``[X, ...]`` (explicit list): strict mode. Yields from designated executors
+  produce ``type='output'``; all other yields produce ``type='intermediate'``.
+
+Prerequisites
+- No external services required.
+"""
+
+
+@executor(id="planner")
+async def planner(messages: list[Message], ctx: WorkflowContext[list[Message], str]) -> None:
+    """Non-designated step: emits an intermediate progress note, then forwards."""
+    prompt = messages[0].text if messages else ""
+    await ctx.yield_output(f"plan: starting work on '{prompt}'")
+    await ctx.send_message(messages)
+
+
+@executor(id="researcher")
+async def researcher(messages: list[Message], ctx: WorkflowContext[list[Message], str]) -> None:
+    """Non-designated step: emits intermediate progress, then forwards."""
+    prompt = messages[0].text if messages else ""
+    await ctx.yield_output(f"research: gathering data for '{prompt}'")
+    await ctx.send_message(messages)
+
+
+@executor(id="answerer")
+async def answerer(messages: list[Message], ctx: WorkflowContext[Never, str]) -> None:
+    """Designated terminal: emits the workflow's final answer."""
+    prompt = messages[0].text if messages else ""
+    await ctx.yield_output(f"final answer to '{prompt}': 42")
+
+
+async def main() -> None:
+    # Build with explicit output_executors=[answerer]. Only `answerer` yields
+    # produce type='output' events; planner and researcher emit type='intermediate'.
+    workflow = (
+        WorkflowBuilder(start_executor=planner, output_executors=[answerer])
+        .add_edge(planner, researcher)
+        .add_edge(researcher, answerer)
+        .build()
+    )
+
+    initial = [Message(role="user", contents=["life, the universe, and everything"])]
+
+    print("=== Streaming events (workflow.run(stream=True)) ===")
+    async for event in workflow.run(initial, stream=True):
+        if event.type == "intermediate":
+            print(f"  [intermediate] {event.executor_id}: {event.data}")
+        elif event.type == "output":
+            print(f"  [output]       {event.executor_id}: {event.data}")
+
+    # WorkflowRunResult.get_outputs() filters to type='output' events, so it
+    # only returns the designated terminal yield.
+    print("\n=== Non-streaming run().get_outputs() ===")
+    result = await workflow.run(initial)
+    print(f"  outputs: {result.get_outputs()}")
+
+    # When the same workflow is wrapped via as_agent(), intermediate events
+    # surface as ``text_reasoning`` content; the terminal event surfaces as
+    # ``text`` content. Existing callers reading ``response.text`` get only
+    # the terminal answer because ``.text`` filters to text content.
+    print("\n=== workflow.as_agent() — intermediate → text_reasoning content ===")
+    agent = workflow.as_agent("planner-agent")
+    response = await agent.run("life, the universe, and everything")
+    print(f"  response.text (terminal only): {response.text!r}")
+    reasoning = " | ".join(
+        c.text for m in response.messages for c in m.contents if c.type == "text_reasoning"
+    )
+    print(f"  reasoning content (intermediates): {reasoning!r}")
+
+    """
+    Sample output:
+
+    === Streaming events (workflow.run(stream=True)) ===
+      [intermediate] planner: plan: starting work on 'life, the universe, and everything'
+      [intermediate] researcher: research: gathering data for 'life, the universe, and everything'
+      [output]       answerer: final answer to 'life, the universe, and everything': 42
+
+    === Non-streaming run().get_outputs() ===
+      outputs: ["final answer to 'life, the universe, and everything': 42"]
+
+    === workflow.as_agent() — intermediate → text_reasoning content ===
+      response.text (terminal only): "final answer to 'life, the universe, and everything': 42"
+      reasoning content (intermediates): "plan: starting work on ... | research: gathering data for ..."
+    """
+
+
+if __name__ == "__main__":
+    asyncio.run(main())