Python: [BREAKING] Standardize orchestration terminal outputs as AgentResponse

[Evan Mattson (moonbox3)]

Motivation and Context

Important

Breaking changes are scoped to the still-experimental agent-framework-orchestrations package. Core agent-framework workflow code is non-breaking.

workflow.as_agent().run(prompt) for SequentialBuilder returned the user's input plus every agent's reply instead of the last agent's answer. The same class of bug affected Concurrent, GroupChat, Magentic, and Handoff: they yielded list[Message] conversation dumps as output events, so as_agent() consumers got a conversation dump instead of an answer.

This PR standardizes the terminal-output shape for every built-in orchestration to AgentResponse (or per-chunk AgentResponseUpdate while streaming). One contract: each orchestration's terminal output event carries the meaningful answer, not a conversation transcript.

Per-orchestration contract

Orchestration	Terminal output event
Sequential (agent terminator)	Last agent's AgentResponse (streaming chunks — the last AgentExecutor itself is the workflow's output executor)
Sequential (custom-executor terminator)	The custom executor's own AgentResponse — see "Custom terminators" below
Concurrent (default aggregator)	AgentResponse with one assistant message per participant (no user prompt prepended)
Concurrent (custom aggregator)	Whatever the aggregator yields
GroupChat	AgentResponse with the orchestrator's completion message
Magentic	AgentResponse with the manager's synthesized final answer
Handoff	No terminal yield — per-agent responses surface as output events as agents speak

SequentialBuilder simplification

SequentialBuilder.build() no longer wraps the chain with a synthetic terminal node (_EndWithConversation removed). The last participant is registered as the workflow's output_executor directly. Two consequences:

1. Custom-executor terminators must yield directly. A custom Executor used as the last participant must call await ctx.yield_output(AgentResponse(messages=[...])). Intermediate custom executors continue to use ctx.send_message(list[Message]) for chaining.
2. Output type uniformity. The terminal output is AgentResponse, not list[Message].

The one in-tree sample using this pattern (samples/03-workflows/orchestrations/sequential_custom_executors.py) and one test were updated.

intermediate_outputs=True on builders continues to work as on main — it flips the workflow's output_executors whitelist off so every executor's yield_output surfaces as a workflow output event. No semantic change.

HIL terminator support

AgentApprovalExecutor now accepts allow_direct_output. When the HIL-wrapped agent is the last participant in SequentialBuilder, the inner approval workflow's response surfaces as the wrapping workflow's terminal output event. A small sibling class _TerminalAgentRequestInfoExecutor yields AgentResponse (instead of AgentExecutorResponse) in the terminator role, so the unwrap happens at the source rather than at the WorkflowExecutor boundary in core.

Behavior changes (orchestrations package)

• workflow.as_agent().run(prompt) returns only the answer for each orchestration, matching the agent contract.
• Sequential / GroupChat / Magentic terminal output: list[Message] (full conversation) → AgentResponse (answer only).
• Concurrent default aggregator: list[Message] (user + per-agent) → AgentResponse (per-agent only).
• Handoff: no synthetic terminal event. Per-agent output events are the end of the stream.
• _EndWithConversation removed from SequentialBuilder; custom-executor terminators must yield_output(AgentResponse) directly.

Core changes (non-breaking)

• WorkflowExecutor handlers annotated WorkflowContext[Any, Any] (was WorkflowContext[Any]) — permissive, lets WorkflowExecutor serve as a workflow's output_executor when allow_direct_output=True.
• WorkflowAgent's streaming converter constructs fresh AgentResponseUpdate instances instead of mutating the executor's payload — independent mutation-safety fix surfaced during review.

Contribution Checklist

• The code builds clean without any errors or warnings
• The PR follows the Contribution Guidelines
• All unit tests pass, and I have added new tests where possible
• Is this a breaking change? If yes, add "[BREAKING]" prefix to the title of the PR.

[Evan Mattson (moonbox3)]

Python Test Coverage Report •

File	Stmts	Miss	Cover	Missing
packages/core/agent_framework/_workflows
_agent.py	358	71	80%	66, 74–80, 116–117, 210, 271, 284, 351, 362, 364, 424, 430, 440–441, 448, 450, 456, 518–519, 528, 536, 562, 595–597, 599, 601, 603, 608, 613, 667, 697, 714, 753–756, 762, 768, 772–773, 776–782, 786–787, 795, 856, 863, 869–870, 881, 913, 920, 941, 950, 954, 956–958, 965
_agent_executor.py	200	16	92%	166, 190, 231, 255, 275–276, 356–358, 360, 370–371, 490–491, 563, 569
_workflow_executor.py	187	30	83%	94, 404, 464, 487, 489, 497–498, 503, 505, 510, 512, 593–599, 603–605, 613, 618, 629, 639, 643, 649, 653, 663, 667
packages/orchestrations/agent_framework_orchestrations
_base_group_chat_orchestrator.py	172	12	93%	109, 277, 292, 326–328, 332, 351, 439, 485–487
_concurrent.py	144	26	81%	51, 60–61, 69–70, 89–90, 95, 113, 118, 123–124, 145, 155, 162, 231, 247, 250, 307, 337, 339–340, 342, 347, 360, 364
_group_chat.py	329	70	78%	175, 344, 351, 384, 395–396, 402, 407, 428, 432, 445–446, 459, 474–475, 477, 493, 520–525, 527, 561–564, 566, 571–575, 664, 667, 706, 709, 712, 715, 723, 735–736, 738–739, 741–742, 744, 749, 752, 761, 767, 811–812, 816–817, 831–832, 834–835, 866–867, 933, 952, 960, 965–967, 974, 984
_handoff.py	335	52	84%	104–105, 107, 160–170, 172, 174, 176, 181, 312, 337, 364, 390, 454, 497, 505, 509–510, 541–543, 548–550, 670, 673, 686, 748, 753, 760, 770, 772, 791, 793, 875–876, 908–909, 1010, 1017, 1089–1090, 1092
_magentic.py	592	91	84%	65–74, 79, 83–94, 259, 270, 274, 294, 355, 364, 366, 408, 425, 434–435, 437–439, 441, 452, 595, 597, 637, 687, 723–725, 727, 737, 745–746, 813–816, 907, 913, 919, 961, 999, 1031, 1048, 1059, 1116–1117, 1121–1123, 1147, 1171–1172, 1185, 1205, 1228, 1273–1274, 1312–1313, 1477, 1480, 1489, 1492, 1497, 1548–1549, 1590–1591, 1639, 1669, 1727, 1741, 1752
_orchestration_request_info.py	66	1	98%	162
_sequential.py	80	6	92%	48, 120, 131, 137, 185, 213
TOTAL	30444	3541	88%

Python Unit Test Overview

Tests	Skipped	Failures	Errors	Time
6122	30 💤	0 ❌	0 🔥	1m 35s ⏱️

[Copilot]

Pull request overview

This PR changes Python orchestration output semantics so workflow.as_agent().run(prompt) returns only the final “answer” (as an AgentResponse) while intermediate participant activity is surfaced via data events (instead of additional output events). It adds a single new knob (AgentExecutor(emit_intermediate_data=...)) and updates orchestrations, tests, and a sample to align with the clarified contract.

Changes:

• Add AgentExecutor.emit_intermediate_data to emit parallel data events for AgentResponse / AgentResponseUpdate.
• Update Sequential/Concurrent/GroupChat/Magentic/Handoff orchestrations to reserve output for terminal answers and use data for intermediate agent activity.
• Update orchestration tests and a sample to validate/illustrate the new terminal vs intermediate event behavior.

Reviewed changes

Copilot reviewed 16 out of 16 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
python/samples/03-workflows/agents/sequential_workflow_as_agent.py	Updates sample to reflect as_agent() returning only the final agent response and notes intermediate observation via data.
python/packages/orchestrations/tests/test_sequential.py	Rewrites sequential tests to assert terminal output is last agent only and intermediate activity is data when enabled.
python/packages/orchestrations/tests/test_magentic.py	Updates Magentic tests to expect AgentResponse terminal outputs and data events for intermediate updates.
python/packages/orchestrations/tests/test_handoff.py	Updates Handoff tests to reflect no synthetic terminal output and per-agent outputs as the stream result.
python/packages/orchestrations/tests/test_group_chat.py	Updates GroupChat tests to expect a single terminal AgentResponse from the orchestrator and intermediate agent updates as data.
python/packages/orchestrations/tests/test_concurrent.py	Updates Concurrent tests to expect default aggregator returns AgentResponse with assistant messages only (no user prompt).
python/packages/orchestrations/agent_framework_orchestrations/_sequential.py	Changes sequential wiring so the last agent executor is the workflow output executor; uses data events for intermediate participants.
python/packages/orchestrations/agent_framework_orchestrations/_orchestration_request_info.py	Plumbs emit_intermediate_data through AgentApprovalExecutor into the inner AgentExecutor.
python/packages/orchestrations/agent_framework_orchestrations/_magentic.py	Changes Magentic terminal output to AgentResponse; adds participant emit_intermediate_data plumbing; restricts output executor to orchestrator.
python/packages/orchestrations/agent_framework_orchestrations/_handoff.py	Removes terminal yield; relies on per-agent output events as the observable result.
python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py	Changes GroupChat completion to yield an AgentResponse; uses participant emit_intermediate_data; restricts outputs to orchestrator.
python/packages/orchestrations/agent_framework_orchestrations/_concurrent.py	Changes default aggregator to yield AgentResponse (assistant replies only) and uses participant emit_intermediate_data for intermediate observation.
python/packages/orchestrations/agent_framework_orchestrations/_base_group_chat_orchestrator.py	Updates base orchestrator termination/max-round outputs to yield AgentResponse completion messages.
python/packages/core/tests/workflow/test_agent_executor.py	Adds core tests asserting emit_intermediate_data produces data events in both streaming and non-streaming modes.
python/packages/core/agent_framework/_workflows/_agent_executor.py	Implements emit_intermediate_data by emitting WorkflowEvent.emit(...)/type='data' alongside existing outputs.
python/packages/core/agent_framework/_workflows/_agent.py	Updates WorkflowAgent to consume data events carrying AgentResponse / AgentResponseUpdate as part of agent responses.

[Evan Mattson (moonbox3)]

Automated Code Review

Reviewers: 3 | Confidence: 82%

✓ Correctness

This PR refactors orchestration output from list[Message] to AgentResponse and introduces emit_intermediate_data on AgentExecutor to surface intermediate participants via data events. The core mechanics are sound: output_executors is now always set (not conditioned on intermediate_outputs), intermediate agents emit data events, and the _agent.py adapter layer correctly processes both output and data events. The existing unresolved review comments remain valid (AgentApprovalExecutor as last participant in sequential, sample file stale imports/docstrings, data-event propagation through WorkflowExecutor, and docstring ordering in concurrent aggregator). I found no additional correctness bugs beyond those already raised.

✓ Security Reliability

This PR refactors orchestration outputs from list[Message] to AgentResponse and introduces emit_intermediate_data on AgentExecutor to surface intermediate participants as data events while reserving output events for the terminal answer. The design is clean and the output_executors / data-event split is well-implemented. The key security/reliability issues already identified in previous review comments (AgentApprovalExecutor as terminal participant producing no output, docstring ordering inaccuracy, data event forwarding in WorkflowExecutor) remain unresolved and still apply. One new concern: the handoff workflow removes all terminal yield_output calls, meaning a termination condition that fires before any agent speaks (e.g., from user input alone) will cause the workflow to go idle with zero output events — consumers relying on at least one output would see silent completion.

✓ Test Coverage

The PR introduces a significant refactoring of orchestration output contracts (from list[Message] to AgentResponse) and adds emit_intermediate_data wiring to surface per-agent responses as data events. Test coverage for sequential, group-chat, and magentic workflows is solid, with good new tests for non-streaming, streaming, intermediate outputs, and as_agent scenarios. However, there are notable gaps: ConcurrentBuilder's intermediate_outputs=True path is completely untested despite new wiring, the handoff async-termination test was weakened to a bare IDLE-state check, and sequential intermediate_outputs is only tested in non-streaming mode.

---

Automated review by moonbox3's agents

[Eduard van Valkenburg (eavanvalkenburg)]

small nit on the env var names, otherwise good to go