Python: Improve the handling of intermediate outputs for workflows and orchestrations (#5623)

* Improve the handling of intermediate outputs for workflows and orchestrations * Address PR review feedback on intermediate output forwarding - Switch workflow.as_agent() forwarding to an explicit allowlist of {output, intermediate, data, request_info} so orchestration-internal events (group_chat, handoff_sent, magentic_orchestrator) stay inside the workflow instead of leaking into agent responses via str(data) coercion. - Stop raising on intermediate AgentResponseUpdate in non-streaming run(); surface the partial as a Message with text_reasoning content. The defensive raise still applies to terminal output events, where Update payloads would corrupt message ordering. - Extend the DevUI workflow-event mapper so intermediate yields wrapping plain strings, Messages, and list[Message] render as visible output items instead of generic completed-trace events. - Add orchestration coverage for GroupChat, Handoff, and Magentic builders (default vs intermediate_outputs=True; structural where end-to-end is heavy). * Lift output-designation policy into a value type Replace the ``Workflow._output_executors`` list and the ``RunnerContext.should_label_as_intermediate`` Protocol method with a single immutable ``OutputDesignation`` value type owned by ``Workflow``. Thread the designation as a parameter through the existing call chain (Runner -> EdgeRunner -> Executor -> WorkflowContext) so ``yield_output`` consults the threaded snapshot directly rather than calling back into the runner context. Removes the ``InProcRunnerContext._workflow`` back-reference and the ``WorkflowBuilder.build()`` assignment that wired it up. Adds the public predicate ``Workflow.is_terminal_executor(executor_id)`` for external observers; ``OutputDesignation`` itself stays package-internal. Key decisions - ``OutputDesignation.designated`` is ``frozenset[str] | None`` -- ``None`` preserves legacy "every yield is type='output'" behavior, any frozenset (including empty) opts into strict mode. The ``DeprecationWarning`` for legacy mode at build time is unchanged. - ``output_designation`` is an optional parameter on ``Runner``, ``EdgeRunner.send_message``, ``EdgeRunner._execute_on_target``, ``Executor.execute``, ``Executor._create_context_for_handler``, and ``WorkflowContext.__init__``. Each defaults to legacy ``OutputDesignation()`` so direct callers (Azure Functions ``CapturingRunnerContext``, ``test_runner`` recording fixtures) keep working without ceremony. - The workflow-level filter in ``_run_core`` reads ``self._output_designation`` live, preserving today's semantics where mutating the designation after build still affects subsequent runs (used by two existing tests). - ``Workflow.to_dict()`` continues to emit ``"output_executors": list[str] | None`` (sorted from the frozenset). Checkpoint format unchanged. Files changed - _workflow.py: add ``OutputDesignation`` dataclass; replace ``_output_executors`` with ``_output_designation``; add ``is_terminal_executor``; delete ``_should_yield_output_event``. - _runner_context.py: drop ``should_label_as_intermediate`` Protocol method and ``InProcRunnerContext`` impl; drop ``_workflow`` back-reference. - _workflow_builder.py: remove ``context._workflow = workflow`` assignment. - _runner.py, _edge_runner.py, _executor.py, _workflow_context.py: thread ``output_designation`` parameter through the call chain. - tests/workflow/test_output_designation.py (new): three-state coverage of the value type plus the public predicate delegation. - tests/workflow/test_workflow_builder.py, test_validation.py, test_workflow.py, test_runner.py and orchestrations/tests/test_orchestration_intermediate_vs_terminal.py: switch probes from ``_output_executors`` set checks to ``get_output_executors`` / ``is_terminal_executor``; update two post-build mutation tests to set ``_output_designation`` instead. Verification - core/tests/workflow/, orchestrations/tests/, azurefunctions/tests/: 1119 passed, 42 skipped, 2 xfailed. - ``uv run poe lint``: clean. - ``uv run poe typing``: only the pre-existing ``_AGENT_FORWARDED_EVENT_TYPES`` pyright warning from 394bcd607 remains. Notes for next iteration - The builder's own ``_output_executors`` attribute (``list[Executor | SupportsAgentRun]``) is intentionally untouched; the issue scoped the rename to the workflow attribute. - Adjacent review candidates (twin ``WorkflowAgent`` translators, ``_AGENT_FORWARDED_EVENT_TYPES`` kind classifier, ``_event_origin_context`` ContextVar removal, ``WorkflowEvent`` ADT split, legacy-mode removal) remain out of scope. * Add explicit workflow output designation Key decisions - Extend the internal OutputDesignation value type from terminal-only membership to output/intermediate/hidden classification. Legacy mode remains outputs=None, so workflows built without output_executors or intermediate_executors still label every yield_output as type='output'. - WorkflowBuilder now accepts intermediate_executors. Providing either designation enters explicit mode; output executors emit output, intermediate executors emit intermediate, and unlisted yield_output payloads are hidden from caller-facing events while remaining in executor_completed data. - Empty explicit designation, duplicate entries, overlaps, unknown executors, and designated executors without workflow output annotations fail build validation. Existing orchestration builders pass intermediate-capable participants through intermediate_executors to preserve current intermediate_outputs behavior until participant-oriented designation lands. Files changed - packages/core/agent_framework/_workflows/_workflow.py, _workflow_builder.py, _workflow_context.py, _validation.py, _events.py - packages/core/tests/workflow/test_output_designation.py, test_output_executors_contract.py, test_strict_mode_event_labeling.py, test_validation.py, test_workflow.py, test_workflow_agent_intermediate.py - packages/orchestrations/agent_framework_orchestrations/_sequential.py, _concurrent.py, _group_chat.py, _magentic.py - packages/core/AGENTS.md Verification - uv run pytest packages/core/tests/workflow packages/orchestrations/tests packages/devui/tests/devui/test_mapper.py -q - uv run pytest packages/azurefunctions/tests -q - uv run poe lint - uv run poe typing fails only on pre-existing packages/core/agent_framework/_workflows/_agent.py _AGENT_FORWARDED_EVENT_TYPES private-use pyright error. Notes for next iteration - issues/03-core-workflow-explicit-designation.md was moved to issues/done but issues/ remains untracked and intentionally excluded from this commit. - Slice 4 should tighten workflow.as_agent() mapping for hidden emissions and streaming-only update payloads; Slice 5 should replace orchestration intermediate_outputs with participant-oriented designation. * Tighten workflow-as-agent output mapping Key decisions - Treat AgentResponseUpdate as a streaming-only payload across the workflow.as_agent() adapter, so non-streaming agent runs now reject both terminal output and intermediate workflow events carrying updates. - Keep streaming classification behavior explicit: terminal update payloads remain normal text content, while intermediate update payloads are rewritten to text_reasoning content. - Add explicit-mode coverage proving hidden yield_output emissions do not appear in non-streaming AgentResponse messages or streaming AgentResponseUpdate chunks. Files changed - packages/core/agent_framework/_workflows/_agent.py - packages/core/tests/workflow/test_workflow_agent_intermediate.py Verification - uv run pytest packages/core/tests/workflow/test_workflow_agent_intermediate.py -q - uv run pytest packages/core/tests/workflow/test_workflow_agent.py packages/core/tests/workflow/test_workflow_agent_intermediate.py -q - uv run pytest packages/core/tests/workflow packages/orchestrations/tests packages/devui/tests/devui/test_mapper.py -q - uv run poe lint - uv run poe typing fails only on the pre-existing packages/core/agent_framework/_workflows/_agent.py _AGENT_FORWARDED_EVENT_TYPES private-use pyright error. Blockers or notes for next iteration - issues/04-workflow-as-agent-output-mapping.md was moved to issues/done/ but issues/ remains untracked and intentionally excluded from this commit. - Slice 5 should replace orchestration intermediate_outputs with participant-oriented designation. * Add orchestration participant output designation Key decisions - Replace orchestration intermediate_outputs with participant-oriented output_participants and intermediate_participants across Sequential, Concurrent, GroupChat, Magentic, and Handoff builders. - Keep synthetic final executors terminal by default for Concurrent, GroupChat, and Magentic; keep Sequential's final participant terminal by default; keep Handoff participants terminal by default. - Centralize participant designation validation for empty explicit designation, duplicates, overlaps, and unknown participants, then map validated participants to workflow output/intermediate executors. Files changed - packages/orchestrations/agent_framework_orchestrations/_participant_designation.py - packages/orchestrations/agent_framework_orchestrations/_sequential.py - packages/orchestrations/agent_framework_orchestrations/_concurrent.py - packages/orchestrations/agent_framework_orchestrations/_group_chat.py - packages/orchestrations/agent_framework_orchestrations/_magentic.py - packages/orchestrations/agent_framework_orchestrations/_handoff.py - packages/orchestrations/tests/test_orchestration_intermediate_vs_terminal.py - packages/orchestrations/tests/test_magentic.py Blockers or notes for next iteration - issues/05-orchestration-participant-designation.md was moved to issues/done/ but issues/ remains untracked and intentionally excluded from this commit. - Slice 7 should migrate samples and docs away from intermediate_outputs to the new participant designation API. - uv run poe typing still fails only on the pre-existing packages/core/agent_framework/_workflows/_agent.py _AGENT_FORWARDED_EVENT_TYPES private-use pyright error. * Migrate samples to explicit output designation Key decisions - Replace sample usage of the removed orchestration intermediate_outputs boolean with participant-oriented intermediate_participants designation. - Update raw workflow guidance to show output_executors together with intermediate_executors, and document that unlisted yields are hidden in explicit designation mode. - Keep orchestration final outputs terminal while streaming designated participant responses as intermediate progress, including workflow.as_agent() samples where intermediates map to text_reasoning content. - Refresh workflow and orchestration README guidance plus the changelog reference so public docs no longer point users at intermediate_outputs. Files changed - CHANGELOG.md - packages/orchestrations/README.md - samples/README.md - samples/03-workflows/README.md - samples/03-workflows/control-flow/intermediate_vs_terminal_outputs.py - samples/03-workflows/orchestrations/README.md - samples/03-workflows/orchestrations/group_chat_agent_manager.py - samples/03-workflows/orchestrations/group_chat_philosophical_debate.py - samples/03-workflows/orchestrations/group_chat_simple_selector.py - samples/03-workflows/orchestrations/magentic.py - samples/03-workflows/orchestrations/magentic_human_plan_review.py - samples/03-workflows/orchestrations/sequential_chain_only_agent_responses.py - samples/03-workflows/agents/group_chat_workflow_as_agent.py - samples/03-workflows/agents/magentic_workflow_as_agent.py - samples/03-workflows/agents/sequential_workflow_as_agent.py - samples/semantic-kernel-migration/orchestrations/group_chat.py - samples/semantic-kernel-migration/orchestrations/magentic.py Blockers or notes for next iteration - issues/07-samples-and-docs-explicit-output-designation.md was moved to issues/done/ but issues/ remains untracked and intentionally excluded from this commit. - issues/06-devui-intermediate-event-rendering.md remains present and appears already satisfied by existing DevUI mapper/tests from the prior implementation slice. - PRD-explicit-workflow-output-designation.md remains untracked and intentionally excluded from this commit. * Render DevUI intermediate workflow outputs Key decisions - Preserve workflow output designation metadata on visible DevUI output messages and text deltas so intermediate/data emissions remain distinguishable from terminal output. - Render intermediate workflow message items in the execution timeline using executor metadata, while excluding them from the final workflow result aggregation. - Keep terminal output message rendering unchanged and retain legacy data events on the intermediate compatibility path. Files changed - packages/devui/agent_framework_devui/_mapper.py - packages/devui/frontend/src/components/features/workflow/execution-timeline.tsx - packages/devui/frontend/src/components/features/workflow/workflow-view.tsx - packages/devui/frontend/src/types/openai.ts - packages/devui/tests/devui/test_mapper.py Blockers or notes for next iteration - issues/06-devui-intermediate-event-rendering.md was moved to issues/done/ but issues/ remains untracked and intentionally excluded from this commit. - PRD-explicit-workflow-output-designation.md remains untracked and intentionally excluded from this commit. - uv run poe typing still fails only on the pre-existing packages/core/agent_framework/_workflows/_agent.py _AGENT_FORWARDED_EVENT_TYPES private-use pyright error. * Fix mypy * Clarify orchestration participant output config * Rename participant output kwargs for clarity output_participants -> final_output_from, intermediate_participants -> intermediate_output_from. The old names read like categories of participant; the new names make it clear the kwarg designates which participants' outputs surface as final vs. intermediate events. * Rename core workflow output kwargs with deprecation shim Adds final_output_from / intermediate_output_from as canonical kwargs on Workflow and WorkflowBuilder. Old output_executors / intermediate_executors kwargs continue to work but emit DeprecationWarning via a shared coalesce helper that also rejects supplying both. Wire-format keys in to_dict() stay as output_executors / intermediate_executors so checkpoint compatibility is preserved. Internal call sites in orchestrations and samples updated to the new names so users following sample code learn the canonical vocabulary; legacy callers still work with a one-shot warning. * Suppress pyright reportPrivateUsage on cross-module sentinel import * Update docstrings * Propagate sub-workflow intermediate outputs, fix handoff/sequential intermediate-only designation, and shore up tests, sample, and docstrings around the intermediate output contract. * Add canonical workflow output_from selection Key decisions:\n- Make output_from the canonical workflow-output allow-list and keep output_executors/final_output_from as deprecated compatibility aliases.\n- Treat empty output_from/intermediate_output_from lists as explicit selections and keep validation responsible for empty, duplicate, overlap, and unknown selections.\n- Remove the branch-only public intermediate_executors WorkflowBuilder kwarg while preserving legacy wire keys in to_dict().\n\nFiles changed:\n- packages/core/agent_framework/_workflows/_workflow.py\n- packages/core/agent_framework/_workflows/_workflow_builder.py\n- packages/core/agent_framework/_workflows/_workflow_context.py\n- packages/core/agent_framework/_workflows/_agent.py\n- packages/core/agent_framework/_workflows/_agent_executor.py\n- packages/core/tests/workflow/* output-selection coverage updates\n- packages/core/AGENTS.md\n- issues/done/001-canonical-list-based-output-selection.md\n\nBlockers/notes:\n- Orchestration builders still pass final_output_from internally; follow-up issue 004 should migrate them to output_from.\n- Legacy omitted-selection behavior and explicit all/all_other literals are left for issues 002 and 003. * Add explicit all workflow output selection Key decisions: - Treat output_from='all' as an explicit workflow-output selection sentinel and expand it at build time to executors with declared workflow output types. - Keep omitted output selections in legacy all-output mode with a deprecation warning that names output_from and intermediate_output_from and points to output_from='all'. - Reject intermediate_output_from='all' at construction because the all-output literal is output-only for this issue. Files changed: - packages/core/agent_framework/_workflows/_workflow_builder.py - packages/core/tests/workflow/test_output_executors_contract.py - issues/done/002-explicit-all-output-and-legacy-migration.md Blockers/notes: - all_other intermediate-output selection remains for issue 003. - Workflow-as-agent/orchestration parity remains for issue 004. * Add all-other intermediate output selection Key decisions: - Treat intermediate_output_from='all_other' as an explicit intermediate-output selection sentinel and expand it at build time after the workflow graph is complete. - Expand all_other to output-capable executors not selected by output_from; omitted or empty output_from selects no workflow outputs, while output_from='all' leaves an empty intermediate selection. - Keep output_from='all_other' invalid so all_other remains intermediate-output-only and runtime classification still receives concrete executor-id sets. Files changed: - packages/core/agent_framework/_workflows/_workflow_builder.py - packages/core/tests/workflow/test_output_executors_contract.py - issues/done/003-all-other-intermediate-output-selection.md Blockers/notes: - Workflow-as-agent and orchestration parity remains for issue 004. - Full documentation updates remain for issue 005. * Add orchestration output selection parity Key decisions: - Expose output_from on sequential, concurrent, group chat, handoff, and magentic builders while keeping final_output_from as a deprecated compatibility alias. - Resolve orchestration participant selections through the same explicit rules as workflows: output_from='all', intermediate_output_from='all_other', hidden unselected participant payloads, and overlap/duplicate/unknown/invalid-literal validation. - Continue preserving documented orchestration defaults by always designating each pattern's terminal internal executor where applicable. Files changed: - packages/orchestrations/agent_framework_orchestrations/_participant_output_config.py - packages/orchestrations/agent_framework_orchestrations/_sequential.py - packages/orchestrations/agent_framework_orchestrations/_concurrent.py - packages/orchestrations/agent_framework_orchestrations/_group_chat.py - packages/orchestrations/agent_framework_orchestrations/_handoff.py - packages/orchestrations/agent_framework_orchestrations/_magentic.py - packages/orchestrations/agent_framework_orchestrations/_orchestration_request_info.py - packages/orchestrations/tests/test_orchestration_intermediate_vs_terminal.py - issues/done/004-workflow-as-agent-and-orchestration-parity.md Blockers/notes: - Full documentation and sample migration wording remains for issue 005. - Existing tests that intentionally use final_output_from now emit the new deprecation warning. * Document workflow output selection contract Key decisions: - Use Workflow Output and Intermediate Output as the developer-facing terms for selected caller-facing emissions. - Document output_from and intermediate_output_from as the canonical API, with output_from as an allow-list and unselected payloads hidden unless explicitly selected as intermediate. - Add scenario and invalid-selection tables for workflow and orchestration docs, including legacy omission warnings, output_from='all', intermediate_output_from='all_other', list selections, invalid literals, overlap, duplicates, unknown selections, and empty explicit selections. - Migrate samples away from final_output_from and output_executors except where compatibility aliases are explicitly documented. Files changed: - packages/core/AGENTS.md - packages/orchestrations/README.md - packages/orchestrations/agent_framework_orchestrations/_handoff.py - packages/orchestrations/agent_framework_orchestrations/_sequential.py - samples/03-workflows/README.md - samples/03-workflows/control-flow/intermediate_vs_terminal_outputs.py - samples/03-workflows/human-in-the-loop/agents_with_approval_requests.py - samples/03-workflows/orchestrations/README.md - samples/04-hosting/foundry-hosted-agents/responses/05_workflows/main.py - scripts/sample_validation/create_dynamic_workflow_executor.py - issues/done/005-document-output-selection-contract.md Blockers/notes: - Direct full Ruff on scripts/sample_validation/create_dynamic_workflow_executor.py still reports pre-existing docstring/print/line-length issues outside this docs migration; syntax-focused checks for changed files pass. - No remaining AFK issue files are present under issues/. * Latest updates * Typing fixes * Cleanup
2026-06-16 21:04:09 +08:00 · 2026-05-19 09:15:25 +09:00
parent 3ebbdb01b4
commit 3bbc81554b
68 changed files with 3480 additions and 325 deletions
@@ -272,9 +272,7 @@ async def test_agent_executor_tool_call_with_approval() -> None:
        tools=[mock_tool_requiring_approval],
    )

-    workflow = (
-        WorkflowBuilder(start_executor=agent, output_executors=[test_executor]).add_edge(agent, test_executor).build()
-    )
+    workflow = WorkflowBuilder(start_executor=agent, output_from=[test_executor]).add_edge(agent, test_executor).build()

    # Act
    events = await workflow.run("Invoke tool requiring approval")
@@ -343,9 +341,7 @@ async def test_agent_executor_parallel_tool_call_with_approval() -> None:
        tools=[mock_tool_requiring_approval],
    )

-    workflow = (
-        WorkflowBuilder(start_executor=agent, output_executors=[test_executor]).add_edge(agent, test_executor).build()
-    )
+    workflow = WorkflowBuilder(start_executor=agent, output_from=[test_executor]).add_edge(agent, test_executor).build()

    # Act
    events = await workflow.run("Invoke tool requiring approval")
@@ -512,9 +508,7 @@ async def test_agent_executor_declaration_only_tool_emits_request_info() -> None
        tools=[declaration_only_tool],
    )

-    workflow = (
-        WorkflowBuilder(start_executor=agent, output_executors=[test_executor]).add_edge(agent, test_executor).build()
-    )
+    workflow = WorkflowBuilder(start_executor=agent, output_from=[test_executor]).add_edge(agent, test_executor).build()

    # Act
    events = await workflow.run("Use the client side tool")
@@ -587,9 +581,7 @@ async def test_agent_executor_parallel_declaration_only_tool_emits_request_info(
        tools=[declaration_only_tool],
    )

-    workflow = (
-        WorkflowBuilder(start_executor=agent, output_executors=[test_executor]).add_edge(agent, test_executor).build()
-    )
+    workflow = WorkflowBuilder(start_executor=agent, output_from=[test_executor]).add_edge(agent, test_executor).build()

    # Act
    events = await workflow.run("Use the client side tool")
@@ -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
@@ -177,7 +177,7 @@ async def test_agent_executor_populates_full_conversation_non_streaming() -> Non
    agent_exec = AgentExecutor(agent, id="agent1-exec")
    capturer = _CaptureFullConversation(id="capture")

-    wf = WorkflowBuilder(start_executor=agent_exec, output_executors=[capturer]).add_edge(agent_exec, capturer).build()
+    wf = WorkflowBuilder(start_executor=agent_exec, output_from=[capturer]).add_edge(agent_exec, capturer).build()

    # Act: use run() to test non-streaming mode
    result = await wf.run("hello world")
@@ -344,7 +344,7 @@ async def test_agent_executor_full_conversation_round_trip_does_not_duplicate_hi
    coordinator = _RoundTripCoordinator(target_agent_id="writer_agent")

    wf = (
-        WorkflowBuilder(start_executor=agent_exec, output_executors=[coordinator])
+        WorkflowBuilder(start_executor=agent_exec, output_from=[coordinator])
        .add_edge(agent_exec, coordinator)
        .add_edge(coordinator, agent_exec)
        .build()
@@ -450,7 +450,7 @@ async def test_run_request_with_full_history_clears_service_session_id() -> None
    coordinator = _FullHistoryReplayCoordinator(id="coord", target_exec=spy_exec)

    wf = (
-        WorkflowBuilder(start_executor=tool_exec, output_executors=[coordinator])
+        WorkflowBuilder(start_executor=tool_exec, output_from=[coordinator])
        .add_edge(tool_exec, coordinator)
        .add_edge(coordinator, spy_exec)
        .build()
@@ -478,7 +478,7 @@ async def test_from_response_preserves_service_session_id() -> None:
    # Simulate a prior run on the spy executor.
    spy_exec._session.service_session_id = "resp_PREVIOUS_RUN"  # pyright: ignore[reportPrivateUsage]

-    wf = WorkflowBuilder(start_executor=tool_exec, output_executors=[spy_exec]).add_edge(tool_exec, spy_exec).build()
+    wf = WorkflowBuilder(start_executor=tool_exec, output_from=[spy_exec]).add_edge(tool_exec, spy_exec).build()

    result = await wf.run("start")
    assert result.get_outputs() is not None
@@ -517,7 +517,7 @@ async def test_with_text_preserves_full_conversation_through_custom_executor() -
    capturer = _CaptureFullConversation(id="capture")

    wf = (
-        WorkflowBuilder(start_executor=agent1, output_executors=[capturer])
+        WorkflowBuilder(start_executor=agent1, output_from=[capturer])
        .add_chain([agent1, agent2, _upper_case_executor, agent3, capturer])
        .build()
    )
@@ -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", executor_id="pipeline", data="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"


 # ---------------------------------------------------------------------------
@@ -0,0 +1,137 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Tests for the ``OutputDesignation`` value type and the ``Workflow.is_terminal_executor``
+public predicate that delegates to it.
+
+The states the value type encodes:
+- Omitted-selection compatibility: ``outputs=None`` -> every executor is terminal.
+- Explicit: disjoint ``outputs`` and ``intermediates`` sets classify listed executors,
+  and hide unlisted executors.
+"""
+
+from __future__ import annotations
+
+import pytest
+from typing_extensions import Never
+
+from agent_framework import (
+    Message,
+    WorkflowBuilder,
+    WorkflowContext,
+    WorkflowValidationError,
+    executor,
+)
+from agent_framework._workflows._runner_context import InProcRunnerContext
+from agent_framework._workflows._workflow import OutputDesignation, Workflow
+
+# ---------------------------------------------------------------------------
+# OutputDesignation value type
+# ---------------------------------------------------------------------------
+
+
+def test_omitted_selection_designation_marks_every_executor_as_terminal() -> None:
+    designation = OutputDesignation()  # designated defaults to None
+    assert designation.outputs is None
+    assert designation.is_terminal("anything")
+    assert designation.is_terminal("else")
+    assert designation.classify("anything") == "output"
+
+
+def test_strict_empty_designation_marks_no_executor_as_terminal() -> None:
+    designation = OutputDesignation(outputs=frozenset())
+    assert designation.outputs == frozenset()
+    assert not designation.is_terminal("anything")
+    assert not designation.is_terminal("else")
+    assert designation.classify("anything") is None
+
+
+def test_strict_designated_set_only_terminal_for_members() -> None:
+    designation = OutputDesignation(outputs=frozenset({"alpha", "beta"}), intermediates=frozenset({"gamma"}))
+    assert designation.is_terminal("alpha")
+    assert designation.is_terminal("beta")
+    assert not designation.is_terminal("gamma")
+    assert designation.is_intermediate("gamma")
+    assert designation.classify("alpha") == "output"
+    assert designation.classify("gamma") == "intermediate"
+    assert designation.classify("delta") is None
+
+
+def test_designation_is_frozen() -> None:
+    from dataclasses import FrozenInstanceError
+
+    designation = OutputDesignation(outputs=frozenset({"alpha"}))
+    with pytest.raises(FrozenInstanceError):
+        designation.outputs = frozenset({"beta"})  # type: ignore[misc]
+
+
+# ---------------------------------------------------------------------------
+# Workflow.is_terminal_executor delegates to the designation
+# ---------------------------------------------------------------------------
+
+
+@executor
+async def _emit_one(messages: list[Message], ctx: WorkflowContext[Never, str]) -> None:
+    await ctx.yield_output("hello")
+
+
+@executor
+async def _downstream(message: str, ctx: WorkflowContext[Never, str]) -> None:
+    await ctx.yield_output("downstream")
+
+
+def test_is_terminal_executor_omitted_selection_returns_true_for_any_id() -> None:
+    """Omitted-selection compatibility behavior: every executor is terminal."""
+    import warnings
+
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", DeprecationWarning)
+        workflow = WorkflowBuilder(start_executor=_emit_one).build()
+    assert workflow.is_terminal_executor(_emit_one.id)
+    assert workflow.is_terminal_executor("anything-else")
+
+
+def test_is_intermediate_executor_explicit_list_returns_true_only_for_designated() -> None:
+    """Explicit mode tracks intermediate-designated executors separately."""
+    workflow = WorkflowBuilder(start_executor=_emit_one, intermediate_output_from=[_emit_one]).build()
+    assert not workflow.is_terminal_executor(_emit_one.id)
+    assert not workflow.is_terminal_executor("nope")
+    assert workflow.is_intermediate_executor(_emit_one.id)
+    assert not workflow.is_intermediate_executor("nope")
+
+
+def test_is_terminal_executor_strict_list_returns_true_only_for_designated() -> None:
+    """Strict mode with a designated list: only listed executors are terminal."""
+    workflow = (
+        WorkflowBuilder(start_executor=_emit_one, output_from=[_emit_one]).add_edge(_emit_one, _downstream).build()
+    )
+    assert workflow.is_terminal_executor(_emit_one.id)
+    assert not workflow.is_terminal_executor(_downstream.id)
+
+
+def test_get_output_executors_throws_when_designation_references_missing_executor() -> None:
+    workflow = Workflow(
+        [],
+        {_emit_one.id: _emit_one},
+        _emit_one,
+        InProcRunnerContext(),
+        "test",
+        output_from=["missing"],
+    )
+
+    with pytest.raises(WorkflowValidationError, match="Output executor 'missing' is not present"):
+        workflow.get_output_executors()
+
+
+def test_get_intermediate_executors_throws_when_designation_references_missing_executor() -> None:
+    workflow = Workflow(
+        [],
+        {_emit_one.id: _emit_one},
+        _emit_one,
+        InProcRunnerContext(),
+        "test",
+        output_from=[],
+        intermediate_output_from=["missing"],
+    )
+
+    with pytest.raises(WorkflowValidationError, match="Intermediate executor 'missing' is not present"):
+        workflow.get_intermediate_executors()
@@ -0,0 +1,287 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Tests for the explicit output/intermediate selection contract on WorkflowBuilder."""
+
+from __future__ import annotations
+
+import warnings
+from typing import Any
+
+import pytest
+from typing_extensions import Never
+
+from agent_framework import (
+    Message,
+    WorkflowBuilder,
+    WorkflowContext,
+    WorkflowValidationError,
+    executor,
+)
+
+
+@executor
+async def _emit_one(messages: list[Message], ctx: WorkflowContext[Never, str]) -> None:
+    await ctx.yield_output("hello")
+
+
+@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 test_designation_unset_emits_deprecation_warning() -> None:
+    """State A: WorkflowBuilder built without explicit designation warns."""
+    with pytest.warns(DeprecationWarning, match="output_from or intermediate_output_from") as warning_info:
+        WorkflowBuilder(start_executor=_emit_one).build()
+    assert str(warning_info[0].message) == (
+        "WorkflowBuilder built without explicit output_from or intermediate_output_from; "
+        "every yield_output produces type='output' for compatibility. Pass output_from='all', "
+        "output_from=[...], or intermediate_output_from=[...] to opt into explicit designation - "
+        "explicit designation will be required in a future version."
+    )
+
+
+@pytest.mark.asyncio
+async def test_designation_unset_preserves_compatibility_all_output_behavior() -> None:
+    """Omitted designation keeps compatibility all-output behavior while warning."""
+    with pytest.warns(DeprecationWarning, match="output_from or intermediate_output_from"):
+        workflow = WorkflowBuilder(start_executor=_start).add_edge(_start, _downstream).build()
+
+    result = await workflow.run([Message(role="user", contents=["hi"])])
+
+    assert result.get_outputs() == ["from-start", "from-downstream"]
+    assert result.get_intermediate_outputs() == []
+
+
+@pytest.mark.asyncio
+async def test_output_from_all_emits_all_outputs_without_omitted_selection_warning() -> None:
+    """Explicit all-output designation emits every executor payload without omitted-selection warning."""
+    with warnings.catch_warnings():
+        warnings.simplefilter("error", DeprecationWarning)
+        workflow = WorkflowBuilder(start_executor=_start, output_from="all").add_edge(_start, _downstream).build()
+
+    result = await workflow.run([Message(role="user", contents=["hi"])])
+
+    assert result.get_outputs() == ["from-start", "from-downstream"]
+    assert result.get_intermediate_outputs() == []
+
+
+@pytest.mark.asyncio
+async def test_output_from_all_with_empty_intermediate_list_is_valid() -> None:
+    """Explicit all-output plus an empty intermediate list is a concrete no-intermediate selection."""
+    with warnings.catch_warnings():
+        warnings.simplefilter("error", DeprecationWarning)
+        workflow = (
+            WorkflowBuilder(start_executor=_start, output_from="all", intermediate_output_from=[])
+            .add_edge(_start, _downstream)
+            .build()
+        )
+
+    result = await workflow.run([Message(role="user", contents=["hi"])])
+
+    assert result.get_outputs() == ["from-start", "from-downstream"]
+    assert result.get_intermediate_outputs() == []
+
+
+@pytest.mark.asyncio
+async def test_intermediate_output_from_all_other_marks_non_outputs_as_intermediate() -> None:
+    """All-other intermediate designation classifies every non-output executor yield as intermediate."""
+    workflow = (
+        WorkflowBuilder(
+            start_executor=_start,
+            output_from=[_downstream],
+            intermediate_output_from="all_other",
+        )
+        .add_edge(_start, _downstream)
+        .build()
+    )
+
+    result = await workflow.run([Message(role="user", contents=["hi"])])
+
+    assert result.get_outputs() == ["from-downstream"]
+    assert result.get_intermediate_outputs() == ["from-start"]
+
+
+@pytest.mark.asyncio
+async def test_all_other_streaming_events_mark_non_outputs_as_intermediate() -> None:
+    """All-other emits intermediate events while streaming, not just in collected results."""
+    workflow = (
+        WorkflowBuilder(
+            start_executor=_start,
+            output_from=[_downstream],
+            intermediate_output_from="all_other",
+        )
+        .add_edge(_start, _downstream)
+        .build()
+    )
+    outputs: list[str] = []
+    intermediates: list[str] = []
+
+    async for event in workflow.run([Message(role="user", contents=["hi"])], stream=True):
+        if event.type == "output":
+            outputs.append(event.data)
+        elif event.type == "intermediate":
+            intermediates.append(event.data)
+
+    assert outputs == ["from-downstream"]
+    assert intermediates == ["from-start"]
+
+
+def test_all_other_expands_to_concrete_intermediate_executor_selection_at_build_time() -> None:
+    """The runner receives concrete executor IDs after all-other expansion."""
+    workflow = (
+        WorkflowBuilder(
+            start_executor=_start,
+            output_from=[_downstream],
+            intermediate_output_from="all_other",
+        )
+        .add_edge(_start, _downstream)
+        .build()
+    )
+
+    assert {executor.id for executor in workflow.get_output_executors()} == {_downstream.id}
+    assert {executor.id for executor in workflow.get_intermediate_executors()} == {_start.id}
+    assert workflow.is_intermediate_executor(_start.id)
+    assert not workflow.is_intermediate_executor(_downstream.id)
+
+
+@pytest.mark.asyncio
+async def test_all_other_with_omitted_output_from_emits_only_intermediate_outputs() -> None:
+    """All-other intermediate designation opts out of omitted-selection all-output behavior."""
+    workflow = (
+        WorkflowBuilder(
+            start_executor=_start,
+            intermediate_output_from="all_other",
+        )
+        .add_edge(_start, _downstream)
+        .build()
+    )
+
+    result = await workflow.run([Message(role="user", contents=["hi"])])
+
+    assert result.get_outputs() == []
+    assert result.get_intermediate_outputs() == ["from-start", "from-downstream"]
+
+
+@pytest.mark.asyncio
+async def test_all_other_with_empty_output_from_emits_only_intermediate_outputs() -> None:
+    """All-other intermediate designation treats an empty output list as selecting no workflow outputs."""
+    workflow = (
+        WorkflowBuilder(
+            start_executor=_start,
+            output_from=[],
+            intermediate_output_from="all_other",
+        )
+        .add_edge(_start, _downstream)
+        .build()
+    )
+
+    result = await workflow.run([Message(role="user", contents=["hi"])])
+
+    assert result.get_outputs() == []
+    assert result.get_intermediate_outputs() == ["from-start", "from-downstream"]
+
+
+@pytest.mark.asyncio
+async def test_all_other_with_output_from_all_expands_to_empty_intermediate_selection() -> None:
+    """All-other is empty when every output-capable executor is already selected as workflow output."""
+    workflow = (
+        WorkflowBuilder(
+            start_executor=_start,
+            output_from="all",
+            intermediate_output_from="all_other",
+        )
+        .add_edge(_start, _downstream)
+        .build()
+    )
+
+    result = await workflow.run([Message(role="user", contents=["hi"])])
+
+    assert result.get_outputs() == ["from-start", "from-downstream"]
+    assert result.get_intermediate_outputs() == []
+
+
+@pytest.mark.asyncio
+async def test_intermediate_output_from_all_routes_every_yield_to_intermediate() -> None:
+    """``intermediate_output_from="all"`` designates every output-capable executor as intermediate."""
+    workflow = (
+        WorkflowBuilder(start_executor=_start, intermediate_output_from="all").add_edge(_start, _downstream).build()
+    )
+
+    result = await workflow.run([Message(role="user", contents=["hi"])])
+
+    assert result.get_outputs() == []
+    assert result.get_intermediate_outputs() == ["from-start", "from-downstream"]
+
+
+def test_output_from_all_other_is_rejected() -> None:
+    """The all-other literal is only valid for intermediate output selection."""
+    with pytest.raises(ValueError, match="output_from.*all_other"):
+        WorkflowBuilder(start_executor=_emit_one, output_from="all_other")  # type: ignore[arg-type]
+
+
+@pytest.mark.parametrize(
+    ("output_from", "intermediate_output_from"),
+    [([_emit_one], None), (None, [_emit_one]), ([], [_emit_one])],
+    ids=["output_list", "intermediate_list", "empty_output_with_intermediate"],
+)
+def test_explicit_designation_with_executor_does_not_warn(output_from, intermediate_output_from) -> None:
+    """State B: any explicit designation with at least one executor opts into explicit mode without warning."""
+    with warnings.catch_warnings():
+        warnings.simplefilter("error", DeprecationWarning)
+        WorkflowBuilder(
+            start_executor=_emit_one,
+            output_from=output_from,
+            intermediate_output_from=intermediate_output_from,
+        ).build()
+
+
+@pytest.mark.parametrize(
+    ("output_from", "intermediate_output_from"),
+    [([], None), (None, []), ([], [])],
+    ids=["empty_output", "empty_intermediate", "both_empty"],
+)
+def test_empty_explicit_designation_fails(output_from, intermediate_output_from) -> None:
+    """State C: explicit mode needs at least one output or intermediate executor."""
+    with pytest.raises(WorkflowValidationError, match="at least one output or intermediate executor"):
+        WorkflowBuilder(
+            start_executor=_emit_one,
+            output_from=output_from,
+            intermediate_output_from=intermediate_output_from,
+        ).build()
+
+
+def test_passing_both_output_executors_and_output_from_raises_type_error() -> None:
+    """State D: supplying a deprecated alias and the canonical kwarg is unambiguous user error."""
+    with pytest.raises(TypeError, match="Cannot pass multiple workflow output selection parameters"):
+        WorkflowBuilder(
+            start_executor=_emit_one,
+            output_executors=[_emit_one],
+            output_from=[_emit_one],
+        )
+
+
+def test_intermediate_executors_builder_parameter_is_not_public() -> None:
+    """The branch-only intermediate_executors builder parameter is not supported."""
+    builder_type: Any = WorkflowBuilder
+    with pytest.raises(TypeError, match="unexpected keyword argument 'intermediate_executors'"):
+        builder_type(
+            start_executor=_emit_one,
+            intermediate_executors=[_emit_one],
+        )
+
+
+def test_final_output_from_builder_parameter_is_not_public() -> None:
+    """The branch-only final_output_from builder parameter is not supported."""
+    builder_type: Any = WorkflowBuilder
+    with pytest.raises(TypeError, match="unexpected keyword argument 'final_output_from'"):
+        builder_type(
+            start_executor=_emit_one,
+            final_output_from=[_emit_one],
+        )
@@ -158,7 +158,9 @@ async def test_runner_run_iteration_preserves_message_order_per_edge_runner() ->
        def __init__(self) -> None:
            self.received: list[int] = []

-        async def send_message(self, message: WorkflowMessage, state: State, ctx: RunnerContext) -> bool:
+        async def send_message(
+            self, message: WorkflowMessage, state: State, ctx: RunnerContext, *args: object, **kwargs: object
+        ) -> bool:
            message_data = message.data
            assert isinstance(message_data, MockMessage)
            self.received.append(message_data.data)
@@ -188,7 +190,9 @@ async def test_runner_run_iteration_delivers_different_edge_runners_concurrently
            self.release = asyncio.Event()
            self.call_count = 0

-        async def send_message(self, message: WorkflowMessage, state: State, ctx: RunnerContext) -> bool:
+        async def send_message(
+            self, message: WorkflowMessage, state: State, ctx: RunnerContext, *args: object, **kwargs: object
+        ) -> bool:
            self.call_count += 1
            self.started.set()
            await self.release.wait()
@@ -199,7 +203,9 @@ async def test_runner_run_iteration_delivers_different_edge_runners_concurrently
            self.probe_completed = asyncio.Event()
            self.call_count = 0

-        async def send_message(self, message: WorkflowMessage, state: State, ctx: RunnerContext) -> bool:
+        async def send_message(
+            self, message: WorkflowMessage, state: State, ctx: RunnerContext, *args: object, **kwargs: object
+        ) -> bool:
            self.call_count += 1
            self.probe_completed.set()
            return True
@@ -766,7 +772,7 @@ async def test_runner_with_pre_loop_events():
    runner = Runner([], {}, state, ctx, "test_name", graph_signature_hash="test_hash")

    # Add an event before running
-    await ctx.add_event(WorkflowEvent.output(executor_id="test_executor", data="pre-loop-output"))
+    await ctx.add_event(WorkflowEvent("output", executor_id="test_executor", data="pre-loop-output"))

    events: list[WorkflowEvent] = []
    async for event in runner.run_until_convergence():
@@ -891,7 +897,7 @@ class ExecutorThatFailsWithEvents(Executor):
        # First emit an output event to the workflow context
        await ctx.yield_output(f"output-before-failure-{message.data}")
        # Add some events directly to the runner context
-        await self._runner_ctx.add_event(WorkflowEvent.output(executor_id=self.id, data="pending-event"))
+        await self._runner_ctx.add_event(WorkflowEvent("output", executor_id=self.id, data="pending-event"))
        # Fail on the specified iteration
        if self._iteration_count >= self._fail_on_iteration:
            raise RuntimeError("Executor failed with pending events")
@@ -799,3 +799,48 @@ def test_comprehensive_edge_groups_workflow_serialization() -> None:
    assert len(fan_in_groups[0]["edges"]) == 2, "FanInEdgeGroup should have 2 edges (from parallel_1 and parallel_2)"
    for single_group in single_groups:
        assert len(single_group["edges"]) == 1, "Each SingleEdgeGroup should have exactly 1 edge"
+
+
+def test_to_dict_preserves_compatibility_wire_keys_for_output_designation() -> None:
+    """to_dict() must emit the compatibility wire keys regardless of the Python kwarg names.
+
+    The Python API renamed ``output_executors`` -> ``output_from`` and
+    uses ``intermediate_output_from`` for intermediate selection, but the serialized
+    dict must keep the old keys so existing checkpoints stay readable. This is a
+    regression guard against accidental renames of the wire format.
+    """
+
+    class _Yielder(Executor):
+        @handler
+        async def handle(self, message: str, ctx: WorkflowContext[str, str]) -> None:
+            await ctx.yield_output(message)
+            await ctx.send_message(message)
+
+    class _Terminal(Executor):
+        @handler
+        async def handle(self, message: str, ctx: WorkflowContext[str, str]) -> None:
+            await ctx.yield_output(f"final: {message}")
+
+    start = _Yielder(id="start")
+    progress = _Yielder(id="progress")
+    final = _Terminal(id="final")
+
+    workflow = (
+        WorkflowBuilder(
+            start_executor=start,
+            output_from=[final],
+            intermediate_output_from=[progress],
+        )
+        .add_edge(start, progress)
+        .add_edge(progress, final)
+        .build()
+    )
+
+    d = workflow.to_dict()
+
+    assert "output_executors" in d, "wire key 'output_executors' must be preserved"
+    assert "intermediate_executors" in d, "wire key 'intermediate_executors' must be preserved"
+    assert "output_from" not in d, "new Python kwarg name must NOT leak into the wire format"
+    assert "intermediate_output_from" not in d, "new Python kwarg name must NOT leak into the wire format"
+    assert d["output_executors"] == ["final"]
+    assert d["intermediate_executors"] == ["progress"]
@@ -0,0 +1,118 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Tests for the runner's explicit output selection event labeling."""
+
+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:
+    """Output-designated executor yields produce type='output' events."""
+    workflow = WorkflowBuilder(start_executor=_start, output_from=[_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 intermediate_events == []
+    assert all(ev.data != "from-downstream" for ev in output_events), "unlisted executor yield is hidden"
+
+
+@pytest.mark.asyncio
+async def test_intermediate_designated_executor_emits_intermediate_events() -> None:
+    """Intermediate-designated executor yields produce type='intermediate' events."""
+    workflow = (
+        WorkflowBuilder(start_executor=_start, intermediate_output_from=[_downstream])
+        .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-downstream"}
+
+
+@pytest.mark.asyncio
+async def test_omitted_selection_keeps_all_yields_as_output() -> None:
+    """Omitted output selection 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 output-designated payloads."""
+    workflow = (
+        WorkflowBuilder(
+            start_executor=_start,
+            output_from=[_downstream],
+            intermediate_output_from=[_start],
+        )
+        .add_edge(_start, _downstream)
+        .build()
+    )
+    result = await workflow.run(_input_msg())
+    assert result.get_outputs() == ["from-downstream"]
+    assert result.get_intermediate_outputs() == ["from-start"]
+
+
+@pytest.mark.asyncio
+async def test_hidden_yields_remain_in_executor_completion_events() -> None:
+    """Hidden yield_output payloads stay available through executor_completed observability."""
+    workflow = WorkflowBuilder(start_executor=_start, output_from=[_downstream]).add_edge(_start, _downstream).build()
+    result = await workflow.run(_input_msg())
+    assert result.get_outputs() == ["from-downstream"]
+    assert result.get_intermediate_outputs() == []
+    assert not any(event.type in {"output", "intermediate"} and event.data == "from-start" for event in result)
+    completed = [event for event in result if event.type == "executor_completed" and event.executor_id == _start.id]
+    assert completed
+    assert completed[0].data == ["downstream", "from-start"]
@@ -617,3 +617,75 @@ async def test_sub_workflow_checkpoint_restore_no_duplicate_requests() -> None:
    # Key assertion: Only the second request should be received, not a duplicate of the first
    assert len(request_events) == 1
    assert request_events[0].data.prompt == "Second request"
+
+
+async def test_sub_workflow_intermediate_outputs_propagate_to_parent() -> None:
+    """A child workflow's intermediate emissions must bubble up through the parent.
+
+    Regression guard for the bug where WorkflowExecutor._process_workflow_result only
+    forwarded result.get_outputs() and silently dropped result.get_intermediate_outputs().
+    The forwarded event must carry the WorkflowExecutor's own id as the source so outer
+    callers don't have to know the child's internal executor layout, and it must keep
+    type='intermediate' regardless of how the parent designates the WorkflowExecutor.
+    """
+
+    class _ProgressEmitter(Executor):
+        def __init__(self) -> None:
+            super().__init__(id="progress_emitter")
+
+        @handler
+        async def run(self, message: str, ctx: WorkflowContext[str, str]) -> None:
+            await ctx.yield_output(f"progress: {message}")
+            await ctx.send_message(message)
+
+    class _Finalizer(Executor):
+        def __init__(self) -> None:
+            super().__init__(id="finalizer")
+
+        @handler
+        async def run(self, message: str, ctx: WorkflowContext[Never, str]) -> None:
+            await ctx.yield_output(f"final: {message}")
+
+    progress = _ProgressEmitter()
+    finalizer = _Finalizer()
+    child = (
+        WorkflowBuilder(
+            start_executor=progress,
+            output_from=[finalizer],
+            intermediate_output_from=[progress],
+        )
+        .add_edge(progress, finalizer)
+        .build()
+    )
+
+    sub = WorkflowExecutor(child, id="sub")
+
+    class _ParentSink(Executor):
+        def __init__(self) -> None:
+            super().__init__(id="parent_sink")
+            self.received: list[str] = []
+
+        @handler
+        async def run(self, message: str, ctx: WorkflowContext[Never, str]) -> None:
+            self.received.append(message)
+            await ctx.yield_output(message)
+
+    sink = _ParentSink()
+    parent = WorkflowBuilder(start_executor=sub, output_from=[sink]).add_edge(sub, sink).build()
+
+    intermediate_events: list[WorkflowEvent[Any]] = []
+    output_events: list[WorkflowEvent[Any]] = []
+    async for event in parent.run("hello", stream=True):
+        if event.type == "intermediate":
+            intermediate_events.append(event)
+        elif event.type == "output":
+            output_events.append(event)
+
+    # The child's intermediate emission bubbled up labeled with the WorkflowExecutor id,
+    # not the child's internal executor id.
+    assert len(intermediate_events) == 1, [(e.executor_id, e.data) for e in intermediate_events]
+    assert intermediate_events[0].executor_id == "sub"
+    assert intermediate_events[0].data == "progress: hello"
+
+    # The parent's own terminal output is unaffected.
+    assert any(e.executor_id == "parent_sink" and e.data == "final: hello" for e in output_events)
@@ -550,12 +550,10 @@ def test_output_validation_with_valid_output_executors():
    executor2 = OutputExecutor(id="executor2")

    # Build workflow with valid output executors
-    workflow = (
-        WorkflowBuilder(start_executor=executor1, output_executors=[executor2]).add_edge(executor1, executor2).build()
-    )
+    workflow = WorkflowBuilder(start_executor=executor1, output_from=[executor2]).add_edge(executor1, executor2).build()

    assert workflow is not None
-    assert workflow._output_executors == ["executor2"]  # pyright: ignore[reportPrivateUsage]
+    assert {ex.id for ex in workflow.get_output_executors()} == {"executor2"}


 def test_output_validation_with_multiple_valid_output_executors():
@@ -565,14 +563,14 @@ def test_output_validation_with_multiple_valid_output_executors():
    executor3 = OutputExecutor(id="executor3")

    workflow = (
-        WorkflowBuilder(start_executor=executor1, output_executors=[executor1, executor3])
+        WorkflowBuilder(start_executor=executor1, output_from=[executor1, executor3])
        .add_edge(executor1, executor2)
        .add_edge(executor2, executor3)
        .build()
    )

    assert workflow is not None
-    assert set(workflow._output_executors) == {"executor1", "executor3"}  # pyright: ignore[reportPrivateUsage]
+    assert {ex.id for ex in workflow.get_output_executors()} == {"executor1", "executor3"}


 def test_output_validation_fails_for_nonexistent_executor():
@@ -598,7 +596,7 @@ def test_output_validation_fails_for_executor_without_output_types():

    with pytest.raises(WorkflowValidationError) as exc_info:
        (
-            WorkflowBuilder(start_executor=executor1, output_executors=[no_output_executor])
+            WorkflowBuilder(start_executor=executor1, output_from=[no_output_executor])
            .add_edge(executor1, no_output_executor)
            .build()
        )
@@ -608,16 +606,77 @@ def test_output_validation_fails_for_executor_without_output_types():
    assert exc_info.value.validation_type == ValidationTypeEnum.OUTPUT_VALIDATION


-def test_output_validation_empty_list_passes():
-    """Test that output validation passes with an empty output executors list."""
+def test_output_validation_empty_explicit_designation_fails():
+    """Test that explicit mode rejects an empty output/intermediate designation."""
    executor1 = OutputExecutor(id="executor1")
    executor2 = OutputExecutor(id="executor2")

-    workflow = WorkflowBuilder(start_executor=executor1, output_executors=[]).add_edge(executor1, executor2).build()
+    with pytest.raises(WorkflowValidationError) as exc_info:
+        WorkflowBuilder(start_executor=executor1, output_from=[]).add_edge(executor1, executor2).build()
+
+    assert "at least one output or intermediate executor" in str(exc_info.value)
+    assert exc_info.value.validation_type == ValidationTypeEnum.OUTPUT_VALIDATION
+
+
+def test_output_validation_with_valid_intermediate_executors():
+    """Test that output validation passes when intermediate executors exist and have output types."""
+    executor1 = OutputExecutor(id="executor1")
+    executor2 = OutputExecutor(id="executor2")
+
+    workflow = (
+        WorkflowBuilder(start_executor=executor1, intermediate_output_from=[executor1])
+        .add_edge(executor1, executor2)
+        .build()
+    )

    assert workflow is not None
-    # All executors are outputs
-    assert workflow._output_executors == ["executor1", "executor2"]  # type: ignore
+    assert {ex.id for ex in workflow.get_intermediate_executors()} == {"executor1"}
+    assert workflow.is_intermediate_executor("executor1")
+    assert not workflow.is_terminal_executor("executor2")
+
+
+def test_output_validation_fails_for_designation_overlap():
+    """Test that an executor cannot be both terminal and intermediate."""
+    executor1 = OutputExecutor(id="executor1")
+
+    with pytest.raises(WorkflowValidationError) as exc_info:
+        WorkflowBuilder(
+            start_executor=executor1,
+            output_from=[executor1],
+            intermediate_output_from=[executor1],
+        ).build()
+
+    assert "both output and intermediate" in str(exc_info.value)
+    assert exc_info.value.validation_type == ValidationTypeEnum.OUTPUT_VALIDATION
+
+
+def test_output_validation_fails_for_duplicate_designation():
+    """Test that duplicate output or intermediate designation entries are rejected."""
+    executor1 = OutputExecutor(id="executor1")
+
+    with pytest.raises(WorkflowValidationError) as exc_info:
+        WorkflowBuilder(start_executor=executor1, output_from=[executor1, executor1]).build()
+
+    assert "Duplicate output executor designation" in str(exc_info.value)
+    assert exc_info.value.validation_type == ValidationTypeEnum.OUTPUT_VALIDATION
+
+
+def test_output_validation_fails_for_unknown_intermediate_executor():
+    """Test that intermediate designation rejects executors outside the workflow graph."""
+    executor1 = OutputExecutor(id="executor1")
+    executor2 = OutputExecutor(id="executor2")
+    missing = OutputExecutor(id="missing")
+
+    with pytest.raises(WorkflowValidationError) as exc_info:
+        (
+            WorkflowBuilder(start_executor=executor1, intermediate_output_from=[missing])
+            .add_edge(executor1, executor2)
+            .build()
+        )
+
+    assert "not present in the workflow graph" in str(exc_info.value)
+    assert "missing" in str(exc_info.value)
+    assert exc_info.value.validation_type == ValidationTypeEnum.OUTPUT_VALIDATION


 def test_output_validation_with_direct_validate_workflow_graph():
@@ -1056,7 +1056,7 @@ class PassthroughExecutor(Executor):


 async def test_output_executors_empty_yields_all_outputs() -> None:
-    """Test that when _output_executors is empty (default), all outputs are yielded."""
+    """Test that omitted output selection yields all outputs for compatibility."""
    # Create executors that each produce different outputs
    executor_a = PassthroughExecutor(id="executor_a", output_value=10)
    executor_b = OutputProducerExecutor(id="executor_b", output_value=20)
@@ -1085,9 +1085,7 @@ async def test_output_executors_filters_outputs_non_streaming() -> None:

    # Build workflow with a -> b
    workflow = (
-        WorkflowBuilder(start_executor=executor_a, output_executors=[executor_b])
-        .add_edge(executor_a, executor_b)
-        .build()
+        WorkflowBuilder(start_executor=executor_a, output_from=[executor_b]).add_edge(executor_a, executor_b).build()
    )

    result = await workflow.run(NumberMessage(data=0))
@@ -1110,9 +1108,7 @@ async def test_output_executors_filters_outputs_streaming() -> None:

    # Build workflow with a -> b
    workflow = (
-        WorkflowBuilder(start_executor=executor_a, output_executors=[executor_a])
-        .add_edge(executor_a, executor_b)
-        .build()
+        WorkflowBuilder(start_executor=executor_a, output_from=[executor_a]).add_edge(executor_a, executor_b).build()
    )

    # Collect outputs from streaming
@@ -1136,7 +1132,7 @@ async def test_output_executors_with_multiple_specified_executors() -> None:

    # Build workflow with a -> b -> c
    workflow = (
-        WorkflowBuilder(start_executor=executor_a, output_executors=[executor_a, executor_c])
+        WorkflowBuilder(start_executor=executor_a, output_from=[executor_a, executor_c])
        .add_edge(executor_a, executor_b)
        .add_edge(executor_b, executor_c)
        .build()
@@ -1154,12 +1150,15 @@ async def test_output_executors_with_multiple_specified_executors() -> None:

 async def test_output_executors_with_nonexistent_executor_id() -> None:
    """Test that specifying a non-existent executor ID doesn't break the workflow."""
+    from agent_framework._workflows._workflow import OutputDesignation
+
    executor_a = OutputProducerExecutor(id="executor_a", output_value=42)

    workflow = WorkflowBuilder(start_executor=executor_a).build()

-    # Set output_executors to an ID that doesn't exist
-    workflow._output_executors = ["nonexistent_executor"]  # type: ignore
+    # Designate a nonexistent executor so the workflow-level filter drops every yield.
+    workflow._output_designation = OutputDesignation(outputs=frozenset({"nonexistent_executor"}))  # type: ignore[attr-defined]
+    workflow._runner.context.set_yield_output_classifier(workflow._output_designation.classify)  # type: ignore[attr-defined,reportPrivateUsage]

    result = await workflow.run(NumberMessage(data=0))
    outputs = result.get_outputs()
@@ -1199,7 +1198,7 @@ async def test_output_executors_filtering_with_fan_in() -> None:

    # Build fan-in workflow: start -> [a, b] -> aggregator
    workflow = (
-        WorkflowBuilder(start_executor=executor_start, output_executors=[aggregator])
+        WorkflowBuilder(start_executor=executor_start, output_from=[aggregator])
        .add_fan_out_edges(executor_start, [executor_a, executor_b])
        .add_fan_in_edges([executor_a, executor_b], aggregator)
        .build()
@@ -1218,7 +1217,7 @@ async def test_output_executors_filtering_with_run_responses() -> None:
    """Test output filtering works correctly with run(responses=...) method."""
    executor = MockExecutorRequestApproval(id="approval_executor")

-    workflow = WorkflowBuilder(start_executor=executor, output_executors=[executor]).build()
+    workflow = WorkflowBuilder(start_executor=executor, output_from=[executor]).build()

    # Run workflow which will request approval
    result = await workflow.run(NumberMessage(data=42))
@@ -1252,8 +1251,11 @@ async def test_output_executors_filtering_with_run_responses_streaming() -> None
    request_events = [e for e in events_list if e.type == "request_info"]
    assert len(request_events) == 1

-    # Set output_executors to exclude the approval executor
-    workflow._output_executors = ["other_executor"]  # type: ignore
+    # Designate a different executor so the workflow-level filter drops the approval yield.
+    from agent_framework._workflows._workflow import OutputDesignation
+
+    workflow._output_designation = OutputDesignation(outputs=frozenset({"other_executor"}))  # type: ignore[attr-defined]
+    workflow._runner.context.set_yield_output_classifier(workflow._output_designation.classify)  # type: ignore[attr-defined,reportPrivateUsage]

    # Send approval response via streaming
    responses = {request_events[0].request_id: ApprovalMessage(approved=True)}
@@ -923,7 +923,7 @@ class TestWorkflowAgent:

        # Build workflow: start -> agent1 (no output) -> agent2 (output visible)
        workflow = (
-            WorkflowBuilder(start_executor=start_exec, output_executors=[start_exec, agent2])
+            WorkflowBuilder(start_executor=start_exec, output_from=[start_exec, agent2])
            .add_edge(start_exec, agent1)
            .add_edge(agent1, agent2)
            .build()
@@ -0,0 +1,353 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Tests for WorkflowAgent forwarding of intermediate workflow events.
+
+Covers:
+- type='intermediate' surfaces as AgentResponseUpdate without content-type rewriting
+- type='data' (compatibility alias via WorkflowEvent.emit) is forwarded
+- 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,
+)
+from agent_framework.exceptions import AgentInvalidRequestException
+
+
+@pytest.mark.asyncio
+async def test_workflow_agent_forwards_intermediate_events_without_content_rewrite() -> None:
+    """An intermediate yield from an intermediate-designated executor surfaces through as_agent
+    as an AgentResponseUpdate carrying its original content type."""
+
+    @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_from=[terminal],
+            intermediate_output_from=[emit],
+        )
+        .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)
+
+    text = " ".join(c.text for u in updates for c in u.contents if c.type == "text")
+    reasoning_text = " ".join(c.text for u in updates for c in u.contents if c.type == "text_reasoning")
+
+    assert "intermediate progress" in text
+    assert "FINAL" in text
+    assert reasoning_text == ""
+
+
+@pytest.mark.asyncio
+async def test_workflow_agent_text_accessor_includes_forwarded_intermediate_text() -> None:
+    """Intermediate text is forwarded as text until issue 5885 defines the final mapping."""
+
+    @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_from=[terminal],
+            intermediate_output_from=[emit],
+        )
+        .add_edge(emit, terminal)
+        .build()
+    )
+    agent = workflow.as_agent("test")
+
+    response = await agent.run("hi")
+    assert isinstance(response, AgentResponse)
+    assert "invisible-progress" in response.text
+    assert "the-answer" in response.text
+
+
+@pytest.mark.asyncio
+async def test_workflow_agent_hidden_yields_do_not_surface_non_streaming() -> None:
+    """In explicit designation mode, unlisted executor yields stay out of agent responses."""
+
+    @executor
+    async def hidden(messages: list[Message], ctx: WorkflowContext[str, str]) -> None:
+        await ctx.yield_output("hidden-progress")
+        await ctx.send_message("forward")
+
+    @executor
+    async def terminal(message: str, ctx: WorkflowContext[Never, str]) -> None:
+        await ctx.yield_output("visible-answer")
+
+    workflow = WorkflowBuilder(start_executor=hidden, output_from=[terminal]).add_edge(hidden, terminal).build()
+    agent = workflow.as_agent("test")
+
+    response = await agent.run("hi")
+    all_text = " ".join(c.text for m in response.messages for c in m.contents if hasattr(c, "text"))
+
+    assert response.text == "visible-answer"
+    assert "hidden-progress" not in all_text
+
+
+@pytest.mark.asyncio
+async def test_workflow_agent_hidden_yields_do_not_surface_streaming() -> None:
+    """In explicit designation mode, unlisted executor yields stay out of agent updates."""
+
+    @executor
+    async def hidden(messages: list[Message], ctx: WorkflowContext[str, str]) -> None:
+        await ctx.yield_output("hidden-progress")
+        await ctx.send_message("forward")
+
+    @executor
+    async def terminal(message: str, ctx: WorkflowContext[Never, str]) -> None:
+        await ctx.yield_output("visible-answer")
+
+    workflow = WorkflowBuilder(start_executor=hidden, output_from=[terminal]).add_edge(hidden, terminal).build()
+    agent = workflow.as_agent("test")
+
+    updates: list[AgentResponseUpdate] = []
+    async for update in agent.run("hi", stream=True):
+        updates.append(update)
+
+    all_text = " ".join(c.text for u in updates for c in u.contents if hasattr(c, "text"))
+
+    assert "visible-answer" in all_text
+    assert "hidden-progress" not in all_text
+
+
+@pytest.mark.asyncio
+async def test_workflow_agent_data_event_emit_factory_still_forwarded() -> None:
+    """Even the deprecated WorkflowEvent.emit() / type='data' path is forwarded."""
+
+    @executor
+    async def emit_data_alias(messages: list[Message], ctx: WorkflowContext[Never, str]) -> None:
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore", DeprecationWarning)
+            await ctx.add_event(WorkflowEvent.emit("emit_data_alias", "data-alias-payload"))
+        await ctx.yield_output("DONE")
+
+    workflow = WorkflowBuilder(start_executor=emit_data_alias, output_from=[emit_data_alias]).build()
+    agent = workflow.as_agent("test")
+
+    updates: list[AgentResponseUpdate] = []
+    async for update in agent.run("hi", stream=True):
+        updates.append(update)
+
+    text = " ".join(c.text for u in updates for c in u.contents if c.type == "text")
+    assert "data-alias-payload" in text
+
+
+@pytest.mark.asyncio
+async def test_workflow_agent_intermediate_message_preserves_additional_properties() -> None:
+    """Message.additional_properties survives intermediate forwarding.
+
+    Producer-attached metadata (tracking_id, conversation_id, etc.) must not disappear
+    for messages flowing through intermediate-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_from=[terminal],
+            intermediate_output_from=[emit],
+        )
+        .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" and c.text == "hi" 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:
+    """A designated executor's text yield surfaces as Content.text."""
+
+    @executor
+    async def only(messages: list[Message], ctx: WorkflowContext[Never, str]) -> None:
+        await ctx.yield_output("the-answer")
+
+    workflow = WorkflowBuilder(start_executor=only, output_from=[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)
+
+
+@pytest.mark.asyncio
+async def test_workflow_agent_non_streaming_rejects_terminal_update() -> None:
+    """A terminal event carrying AgentResponseUpdate is streaming-only and invalid in run()."""
+
+    @executor
+    async def emit(messages: list[Message], ctx: WorkflowContext[Never, AgentResponseUpdate]) -> None:
+        await ctx.yield_output(AgentResponseUpdate(contents=[Content.from_text(text="partial")], role="assistant"))
+
+    workflow = WorkflowBuilder(start_executor=emit, output_from=[emit]).build()
+    agent = workflow.as_agent("test")
+
+    with pytest.raises(AgentInvalidRequestException, match="AgentResponseUpdate"):
+        await agent.run("hi")
+
+
+@pytest.mark.asyncio
+async def test_workflow_agent_non_streaming_rejects_intermediate_update() -> None:
+    """An intermediate event carrying AgentResponseUpdate is streaming-only and invalid in run()."""
+
+    @executor
+    async def emit(messages: list[Message], ctx: WorkflowContext[str, AgentResponseUpdate]) -> None:
+        await ctx.yield_output(AgentResponseUpdate(contents=[Content.from_text(text="partial")], role="assistant"))
+        await ctx.send_message("forward")
+
+    @executor
+    async def terminal(message: str, ctx: WorkflowContext[Never, str]) -> None:
+        await ctx.yield_output("FINAL")
+
+    workflow = (
+        WorkflowBuilder(
+            start_executor=emit,
+            output_from=[terminal],
+            intermediate_output_from=[emit],
+        )
+        .add_edge(emit, terminal)
+        .build()
+    )
+    agent = workflow.as_agent("test")
+
+    with pytest.raises(AgentInvalidRequestException, match="AgentResponseUpdate"):
+        await agent.run("hi")
+
+
+@pytest.mark.asyncio
+async def test_workflow_agent_streaming_update_payloads_preserve_classification() -> None:
+    """Streaming AgentResponseUpdate payloads preserve original content types."""
+
+    @executor
+    async def emit(messages: list[Message], ctx: WorkflowContext[str, AgentResponseUpdate]) -> None:
+        await ctx.yield_output(
+            AgentResponseUpdate(contents=[Content.from_text(text="intermediate-chunk")], role="assistant")
+        )
+        await ctx.send_message("forward")
+
+    @executor
+    async def terminal(message: str, ctx: WorkflowContext[Never, AgentResponseUpdate]) -> None:
+        await ctx.yield_output(
+            AgentResponseUpdate(contents=[Content.from_text(text="terminal-chunk")], role="assistant")
+        )
+
+    workflow = (
+        WorkflowBuilder(
+            start_executor=emit,
+            output_from=[terminal],
+            intermediate_output_from=[emit],
+        )
+        .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)
+
+    text = " ".join(c.text for u in updates for c in u.contents if c.type == "text")
+    reasoning_text = " ".join(c.text for u in updates for c in u.contents if c.type == "text_reasoning")
+
+    assert "intermediate-chunk" in text
+    assert "terminal-chunk" in text
+    assert reasoning_text == ""
+
+
+@pytest.mark.asyncio
+async def test_workflow_agent_drops_orchestration_internal_events() -> None:
+    """Orchestration-internal event types (group_chat / handoff_sent / magentic_orchestrator)
+    must not surface through workflow.as_agent(). Their dataclass payloads would otherwise
+    be stringified by the generic fallback path and leak into response history."""
+
+    @executor
+    async def emit(messages: list[Message], ctx: WorkflowContext[Never, str]) -> None:
+        # Construct typed orchestration-internal events directly to assert they get
+        # dropped at the agent boundary regardless of payload.
+        await ctx.add_event(WorkflowEvent("group_chat", data={"orchestrator": "details"}))  # type: ignore[arg-type]
+        await ctx.add_event(WorkflowEvent("handoff_sent", data={"target": "agent_b"}))  # type: ignore[arg-type]
+        await ctx.add_event(WorkflowEvent("magentic_orchestrator", data={"plan": "..."}))  # type: ignore[arg-type]
+        await ctx.yield_output("FINAL")
+
+    workflow = WorkflowBuilder(start_executor=emit, output_from=[emit]).build()
+    agent = workflow.as_agent("test")
+
+    response = await agent.run("hi")
+    all_text = " ".join(c.text for m in response.messages for c in m.contents if hasattr(c, "text"))
+    assert "orchestrator" not in all_text
+    assert "agent_b" not in all_text
+    assert "plan" not in all_text
+    assert response.text == "FINAL"
+
+
+@pytest.mark.asyncio
+async def test_workflow_agent_drops_orchestration_internal_events_streaming() -> None:
+    """Streaming counterpart — orchestration-internal events stay inside the workflow."""
+
+    @executor
+    async def emit(messages: list[Message], ctx: WorkflowContext[Never, str]) -> None:
+        await ctx.add_event(WorkflowEvent("group_chat", data={"orchestrator": "details"}))  # type: ignore[arg-type]
+        await ctx.yield_output("FINAL")
+
+    workflow = WorkflowBuilder(start_executor=emit, output_from=[emit]).build()
+    agent = workflow.as_agent("test")
+
+    updates: list[AgentResponseUpdate] = []
+    async for update in agent.run("hi", stream=True):
+        updates.append(update)
+
+    all_text = " ".join(c.text for u in updates for c in u.contents if hasattr(c, "text"))
+    assert "orchestrator" not in all_text
+    assert "FINAL" in all_text
@@ -254,10 +254,10 @@ def test_switch_case_with_agents():
 def test_with_output_from_returns_builder():
    """Test that with_output_from returns the builder for method chaining."""
    executor_a = MockExecutor(id="executor_a")
-    builder = WorkflowBuilder(output_executors=[executor_a], start_executor=executor_a)
+    builder = WorkflowBuilder(output_from=[executor_a], start_executor=executor_a)

-    # Verify builder was created with output_executors
-    assert builder._output_executors == [executor_a]  # pyright: ignore[reportPrivateUsage]
+    # Verify builder was created with output_from
+    assert builder._output_from == [executor_a]  # pyright: ignore[reportPrivateUsage]


 def test_with_output_from_with_executor_instances():
@@ -266,13 +266,11 @@ def test_with_output_from_with_executor_instances():
    executor_b = MockExecutor(id="executor_b")

    workflow = (
-        WorkflowBuilder(start_executor=executor_a, output_executors=[executor_b])
-        .add_edge(executor_a, executor_b)
-        .build()
+        WorkflowBuilder(start_executor=executor_a, output_from=[executor_b]).add_edge(executor_a, executor_b).build()
    )

    # Verify that the workflow was built with the correct output executors
-    assert workflow._output_executors == ["executor_b"]  # type: ignore
+    assert {ex.id for ex in workflow.get_output_executors()} == {"executor_b"}


 def test_with_output_from_with_agent_instances():
@@ -280,10 +278,10 @@ def test_with_output_from_with_agent_instances():
    agent_a = DummyAgent(id="agent_a", name="writer")
    agent_b = DummyAgent(id="agent_b", name="reviewer")

-    workflow = WorkflowBuilder(start_executor=agent_a, output_executors=[agent_b]).add_edge(agent_a, agent_b).build()
+    workflow = WorkflowBuilder(start_executor=agent_a, output_from=[agent_b]).add_edge(agent_a, agent_b).build()

    # Verify that the workflow was built with the agent's name as output executor
-    assert workflow._output_executors == ["reviewer"]  # type: ignore
+    assert {ex.id for ex in workflow.get_output_executors()} == {"reviewer"}


 def test_with_output_from_with_executor_instances_by_id():
@@ -292,12 +290,10 @@ def test_with_output_from_with_executor_instances_by_id():
    executor_b = MockExecutor(id="ExecutorB")

    workflow = (
-        WorkflowBuilder(start_executor=executor_a, output_executors=[executor_b])
-        .add_edge(executor_a, executor_b)
-        .build()
+        WorkflowBuilder(start_executor=executor_a, output_from=[executor_b]).add_edge(executor_a, executor_b).build()
    )

-    assert workflow._output_executors == ["ExecutorB"]  # type: ignore
+    assert {ex.id for ex in workflow.get_output_executors()} == {"ExecutorB"}


 def test_with_output_from_with_multiple_executors():
@@ -307,29 +303,27 @@ def test_with_output_from_with_multiple_executors():
    executor_c = MockExecutor(id="executor_c")

    workflow = (
-        WorkflowBuilder(start_executor=executor_a, output_executors=[executor_a, executor_c])
+        WorkflowBuilder(start_executor=executor_a, output_from=[executor_a, executor_c])
        .add_edge(executor_a, executor_b)
        .add_edge(executor_b, executor_c)
        .build()
    )

    # Verify that the workflow was built with both output executors
-    assert set(workflow._output_executors) == {"executor_a", "executor_c"}  # type: ignore
+    assert {ex.id for ex in workflow.get_output_executors()} == {"executor_a", "executor_c"}


 def test_with_output_from_can_be_set_to_different_value():
-    """Test that output_executors can be set at construction time."""
+    """Test that output_from can be set at construction time."""
    executor_a = MockExecutor(id="executor_a")
    executor_b = MockExecutor(id="executor_b")

    workflow = (
-        WorkflowBuilder(start_executor=executor_a, output_executors=[executor_b])
-        .add_edge(executor_a, executor_b)
-        .build()
+        WorkflowBuilder(start_executor=executor_a, output_from=[executor_b]).add_edge(executor_a, executor_b).build()
    )

    # Verify that the setting is applied
-    assert workflow._output_executors == ["executor_b"]  # type: ignore
+    assert {ex.id for ex in workflow.get_output_executors()} == {"executor_b"}


 def test_with_output_from_with_agent_instances_resolves_name():
@@ -338,37 +332,37 @@ def test_with_output_from_with_agent_instances_resolves_name():
    agent_reviewer = DummyAgent(id="agent2", name="reviewer")

    workflow = (
-        WorkflowBuilder(start_executor=agent_writer, output_executors=[agent_reviewer])
+        WorkflowBuilder(start_executor=agent_writer, output_from=[agent_reviewer])
        .add_edge(agent_writer, agent_reviewer)
        .build()
    )

-    assert workflow._output_executors == ["reviewer"]  # type: ignore
+    assert {ex.id for ex in workflow.get_output_executors()} == {"reviewer"}


 def test_with_output_from_in_constructor():
-    """Test that output_executors works correctly when set in the constructor."""
+    """Test that output_from works correctly when set in the constructor."""
    executor_a = MockExecutor(id="executor_a")
    executor_b = MockExecutor(id="executor_b")
    executor_c = MockExecutor(id="executor_c")

-    # Build workflow with output_executors in the constructor
+    # Build workflow with output_from in the constructor
    workflow = (
-        WorkflowBuilder(start_executor=executor_a, output_executors=[executor_c])
+        WorkflowBuilder(start_executor=executor_a, output_from=[executor_c])
        .add_edge(executor_a, executor_b)
        .add_edge(executor_b, executor_c)
        .build()
    )

    # Verify that the setting persists through the chain
-    assert workflow._output_executors == ["executor_c"]  # type: ignore
+    assert {ex.id for ex in workflow.get_output_executors()} == {"executor_c"}


 def test_with_output_from_with_invalid_executor_raises_validation_error():
    """Test that with_output_from with an invalid executor raises an error."""
    executor_a = MockExecutor(id="executor_a")

-    builder = WorkflowBuilder(start_executor=executor_a, output_executors=[MockExecutor(id="executor_b")])
+    builder = WorkflowBuilder(start_executor=executor_a, output_from=[MockExecutor(id="executor_b")])

    # Attempting to set output from an executor not in the workflow should raise an error
    with pytest.raises(
@@ -5,6 +5,7 @@ from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from typing import TYPE_CHECKING, Any

+import pytest
 from typing_extensions import Never

 from agent_framework import (
@@ -72,6 +73,31 @@ async def test_executor_cannot_emit_framework_lifecycle_event(caplog: "LogCaptur
        assert any("attempted to emit" in message and "'status'" in message for message in list(caplog.messages))


+@pytest.mark.parametrize(
+    "event",
+    [
+        WorkflowEvent("output", executor_id="exec", data="output-payload"),
+        WorkflowEvent("intermediate", executor_id="exec", data="intermediate-payload"),
+    ],
+)
+async def test_executor_cannot_emit_output_selection_events(
+    event: WorkflowEvent[Any],
+    caplog: "LogCaptureFixture",
+) -> None:
+    async with make_context() as (ctx, runner_ctx):
+        caplog.clear()
+        with caplog.at_level("WARNING"):
+            await ctx.add_event(event)
+
+        events: list[WorkflowEvent] = await runner_ctx.drain_events()
+        assert len(events) == 1
+        assert events[0].type == "warning"
+        data = events[0].data
+        assert isinstance(data, str)
+        assert "reserved for ctx.yield_output()" in data
+        assert event.data not in [emitted.data for emitted in events]
+
+
 async def test_executor_emits_normal_event() -> None:
    async with make_context() as (ctx, runner_ctx):
        # Create a normal event to test event emission
@@ -0,0 +1,34 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Tests for WorkflowEvent factory methods and WorkflowEvent.emit() deprecation."""
+
+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_output_selection_factories_are_not_public() -> None:
+    """Callers should use ctx.yield_output(), not direct output/intermediate factories."""
+    assert not hasattr(WorkflowEvent, "output")
+    assert not hasattr(WorkflowEvent, "intermediate")
+
+
+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="yield_output"):
+        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"
@@ -377,7 +377,7 @@ async def test_kwargs_preserved_on_response_continuation() -> None:
    from agent_framework import WorkflowBuilder

    agent = _ApprovalCapturingAgent()
-    workflow = WorkflowBuilder(start_executor=agent, output_executors=[agent]).build()
+    workflow = WorkflowBuilder(start_executor=agent, output_from=[agent]).build()

    # Initial run with function_invocation_kwargs — workflow should pause for approval
    fi_kwargs = {"token": "abc"}