From 66ef1cfa840a4267a86bbc04a601d073877f594a Mon Sep 17 00:00:00 2001 From: Giles Odigwe Date: Thu, 6 Aug 2026 15:26:53 -0700 Subject: [PATCH 1/3] Python: Fix group chat invoking a re-selected participant with no messages When a group chat orchestrator selected the participant that had just spoken, that participant was invoked with an empty message list: the latest messages are deliberately not broadcast back to it, and its AgentExecutor cache is cleared after every run. Most chat agents tolerate this (AgentExecutor only logs a warning), but agents that require input reject it outright. A2AAgent raises "At least one message is required when starting a new task", which aborted the whole workflow. Both GroupChatOrchestrator and AgentBasedGroupChatOrchestrator now pass a continuation instruction when the next speaker is the participant that just responded, so the request is never empty. Every other selection is unchanged. MagenticOrchestrator was already unaffected because it always supplies an instruction. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 149dd7ec-b2cc-4108-87bc-df347061932b --- .../_group_chat.py | 11 +- .../orchestrations/tests/test_group_chat.py | 143 ++++++++++++++++++ 2 files changed, 153 insertions(+), 1 deletion(-) diff --git a/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py b/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py index bbb61edae28..3a304372f61 100644 --- a/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py +++ b/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py @@ -69,6 +69,12 @@ logger = logging.getLogger(__name__) +# Sent when a participant is selected to speak again immediately after it spoke: it gets no +# broadcast (its own reply is already in its session) and its executor cache was cleared after +# the previous run, so its request would otherwise carry no messages at all. Some agents +# (for example A2AAgent) reject empty input. +_CONSECUTIVE_TURN_DEFAULT_INSTRUCTION = "Continue the conversation." + @dataclass(frozen=True) class GroupChatState: @@ -233,6 +239,7 @@ async def _handle_response( await self._send_request_to_participant( next_speaker, cast(WorkflowContext[AgentExecutorRequest | GroupChatRequestMessage], ctx), + additional_instruction=_CONSECUTIVE_TURN_DEFAULT_INSTRUCTION if next_speaker == participant else None, ) self._increment_round() @@ -409,10 +416,12 @@ async def _handle_response( participants=[p for p in self._participant_registry.participants if p != participant], ) # Send request to selected participant + next_speaker = agent_orchestration_output.next_speaker await self._send_request_to_participant( # If not terminating, next_speaker must be provided thus will not be None - agent_orchestration_output.next_speaker, # type: ignore[arg-type] + next_speaker, # type: ignore[arg-type] cast(WorkflowContext[AgentExecutorRequest | GroupChatRequestMessage], ctx), + additional_instruction=_CONSECUTIVE_TURN_DEFAULT_INSTRUCTION if next_speaker == participant else None, ) self._increment_round() diff --git a/python/packages/orchestrations/tests/test_group_chat.py b/python/packages/orchestrations/tests/test_group_chat.py index 935111ccaad..8c7c36b4c26 100644 --- a/python/packages/orchestrations/tests/test_group_chat.py +++ b/python/packages/orchestrations/tests/test_group_chat.py @@ -1,5 +1,6 @@ # Copyright (c) Microsoft. All rights reserved. +import json from collections.abc import AsyncIterable, Callable, Sequence from typing import Any, cast @@ -30,6 +31,9 @@ ) from agent_framework_orchestrations import BaseGroupChatOrchestrator +from agent_framework_orchestrations._group_chat import ( # pyright: ignore[reportPrivateUsage] + _CONSECUTIVE_TURN_DEFAULT_INSTRUCTION, +) class StubAgent(BaseAgent): @@ -1098,3 +1102,142 @@ def invalid_factory() -> Any: # endregion + +# region Empty-input regression (issue #7456) + + +class RecordingStubAgent(BaseAgent): + """Stub agent that records received messages and rejects empty input, like ``A2AAgent``.""" + + def __init__(self, agent_name: str, reply_text: str, **kwargs: Any) -> None: + super().__init__(name=agent_name, description=f"Recording stub agent {agent_name}", **kwargs) + self._reply_text = reply_text + self.received_messages: list[list[Message]] = [] + + def run( # type: ignore[override] + self, + messages: str | Content | Message | Sequence[str | Content | Message] | None = None, + *, + stream: bool = False, + session: AgentSession | None = None, + **kwargs: Any, + ) -> Any: + normalized = list(cast(Sequence[Message], messages)) if messages else [] + self.received_messages.append(normalized) + if not normalized: + raise ValueError("At least one message is required when starting a new task (no continuation_token).") + if stream: + return self._run_stream_impl() + return self._run_impl() + + async def _run_impl(self) -> AgentResponse[Any]: + response = Message(role="assistant", contents=[self._reply_text], author_name=self.name) + return AgentResponse(messages=[response]) + + async def _run_stream_impl(self) -> AsyncIterable[AgentResponseUpdate]: + yield AgentResponseUpdate( + contents=[Content.from_text(text=self._reply_text)], role="assistant", author_name=self.name + ) + + +class RepeatSpeakerManagerAgent(Agent): + """Manager that selects the same participant twice in a row, then terminates.""" + + def __init__(self, speaker: str, selections: int = 2) -> None: + super().__init__(client=cast(Any, MockChatClient()), name="manager_agent", description="Repeat manager") + self._speaker = speaker + self._selections = selections + self._call_count = 0 + + async def run( # type: ignore[override] # ty: ignore[invalid-method-override] + self, + messages: str | Content | Message | Sequence[str | Content | Message] | None = None, + *, + session: AgentSession | None = None, + **kwargs: Any, + ) -> AgentResponse[Any]: + if self._call_count < self._selections: + self._call_count += 1 + payload: dict[str, Any] = { + "terminate": False, + "reason": "Selecting agent", + "next_speaker": self._speaker, + "final_message": None, + } + else: + payload = { + "terminate": True, + "reason": "Task complete", + "next_speaker": None, + "final_message": "manager final", + } + return AgentResponse[Any]( + messages=[Message(role="assistant", contents=[json.dumps(payload)], author_name=self.name)], + value=payload, + ) + + +@pytest.mark.parametrize("stream", [False, True]) +async def test_group_chat_consecutive_selection_sends_non_empty_messages(stream: bool) -> None: + """A participant re-selected right after it spoke must not be invoked with empty input.""" + agent = RecordingStubAgent("solo", "reply from solo") + + workflow = GroupChatBuilder( + participants=[agent], + max_rounds=2, + selection_func=lambda state: "solo", + ).build() + + if stream: + async for _ in workflow.run("kickoff", stream=True): + pass + else: + await workflow.run("kickoff") + + assert len(agent.received_messages) == 2, "Expected the participant to be invoked twice" + assert all(received for received in agent.received_messages), "Participant was invoked with empty messages" + assert any( + _CONSECUTIVE_TURN_DEFAULT_INSTRUCTION in (message.text or "") for message in agent.received_messages[1] + ), "Second invocation should carry the continuation instruction" + + +async def test_agent_orchestrator_consecutive_selection_sends_non_empty_messages() -> None: + """Same guarantee when an orchestrator agent (LLM) re-selects the participant that just spoke.""" + agent = RecordingStubAgent("solo", "reply from solo") + + workflow = GroupChatBuilder( + participants=[agent], + orchestrator_agent=RepeatSpeakerManagerAgent("solo"), + ).build() + + async for _ in workflow.run("kickoff", stream=True): + pass + + assert len(agent.received_messages) == 2, "Expected the participant to be invoked twice" + assert all(received for received in agent.received_messages), "Participant was invoked with empty messages" + assert any( + _CONSECUTIVE_TURN_DEFAULT_INSTRUCTION in (message.text or "") for message in agent.received_messages[1] + ), "Second invocation should carry the continuation instruction" + + +async def test_group_chat_alternating_selection_has_no_continuation_instruction() -> None: + """When a different participant speaks each round, no continuation instruction is added.""" + alpha = RecordingStubAgent("alpha", "reply from alpha") + beta = RecordingStubAgent("beta", "reply from beta") + + workflow = GroupChatBuilder( + participants=[alpha, beta], + max_rounds=2, + selection_func=make_sequence_selector(), + ).build() + + async for _ in workflow.run("kickoff", stream=True): + pass + + for participant in (alpha, beta): + for received in participant.received_messages: + assert received, "Participant was invoked with empty messages" + assert all(_CONSECUTIVE_TURN_DEFAULT_INSTRUCTION not in (message.text or "") for message in received) + + +# endregion From cefce977039599c1cf420ba35190be0a0468e670 Mon Sep 17 00:00:00 2001 From: Giles Odigwe Date: Thu, 6 Aug 2026 18:07:50 -0700 Subject: [PATCH 2/3] Python: Normalize stub agent input with the shared workflow helper RecordingStubAgent normalized its input with list(cast(...)), which does not honor its declared messages union: a bare str would silently expand into one Message per character, and a single Message would not iterate as intended. Use normalize_messages_input, the same helper AgentExecutor uses, so the stub normalizes exactly like production and no second implementation can drift. Addresses PR review feedback. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 149dd7ec-b2cc-4108-87bc-df347061932b --- python/packages/orchestrations/tests/test_group_chat.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/python/packages/orchestrations/tests/test_group_chat.py b/python/packages/orchestrations/tests/test_group_chat.py index 8c7c36b4c26..8436e97e454 100644 --- a/python/packages/orchestrations/tests/test_group_chat.py +++ b/python/packages/orchestrations/tests/test_group_chat.py @@ -20,6 +20,7 @@ WorkflowRunState, ) from agent_framework._workflows._checkpoint import InMemoryCheckpointStorage +from agent_framework._workflows._message_utils import normalize_messages_input from agent_framework.orchestrations import ( AgentRequestInfoResponse, GroupChatBuilder, @@ -1122,7 +1123,7 @@ def run( # type: ignore[override] session: AgentSession | None = None, **kwargs: Any, ) -> Any: - normalized = list(cast(Sequence[Message], messages)) if messages else [] + normalized = normalize_messages_input(messages) self.received_messages.append(normalized) if not normalized: raise ValueError("At least one message is required when starting a new task (no continuation_token).") From da6305aa0b7785c6cda4f714102403c80c285105 Mon Sep 17 00:00:00 2001 From: Giles Odigwe Date: Fri, 7 Aug 2026 08:58:54 -0700 Subject: [PATCH 3/3] Python: Also send a continuation instruction when the broadcast is empty The previous guard only covered the participant that had just spoken, but a participant can also end up with an empty cache when it is not the last speaker: clean_conversation_for_handoff drops messages that have no text content, so a tool-only response cleans to an empty list and the broadcast carries nothing. With participants A and B, A speaks (cache cleared), B responds with only tool content, and the orchestrator then selects A. A is not the last speaker, so the old condition skipped the instruction and A ran with an empty cache. Send the instruction when the selected speaker just spoke or when the broadcast itself was empty. Together these cover every case: a non-empty broadcast always reaches every participant except the one that just spoke. Rename the constant to _CONTINUATION_DEFAULT_INSTRUCTION since it is no longer specific to consecutive turns. The same defect exists in the handoff orchestrator and is tracked separately in #7573. Addresses PR review feedback. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 149dd7ec-b2cc-4108-87bc-df347061932b --- .../_group_chat.py | 19 +++-- .../orchestrations/tests/test_group_chat.py | 70 ++++++++++++++++--- 2 files changed, 74 insertions(+), 15 deletions(-) diff --git a/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py b/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py index 3a304372f61..525629c7cf5 100644 --- a/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py +++ b/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py @@ -69,11 +69,12 @@ logger = logging.getLogger(__name__) -# Sent when a participant is selected to speak again immediately after it spoke: it gets no -# broadcast (its own reply is already in its session) and its executor cache was cleared after -# the previous run, so its request would otherwise carry no messages at all. Some agents -# (for example A2AAgent) reject empty input. -_CONSECUTIVE_TURN_DEFAULT_INSTRUCTION = "Continue the conversation." +# Sent when the selected speaker would otherwise receive no messages at all: either it just +# spoke (it gets no broadcast, since its own reply is already in its session) or the broadcast +# itself was empty because cleaning stripped every message. Its executor cache is cleared after +# each run, so in both cases the request would carry nothing and some agents (for example +# A2AAgent) reject empty input. +_CONTINUATION_DEFAULT_INSTRUCTION = "Continue the conversation." @dataclass(frozen=True) @@ -239,7 +240,9 @@ async def _handle_response( await self._send_request_to_participant( next_speaker, cast(WorkflowContext[AgentExecutorRequest | GroupChatRequestMessage], ctx), - additional_instruction=_CONSECUTIVE_TURN_DEFAULT_INSTRUCTION if next_speaker == participant else None, + additional_instruction=( + _CONTINUATION_DEFAULT_INSTRUCTION if next_speaker == participant or not messages else None + ), ) self._increment_round() @@ -421,7 +424,9 @@ async def _handle_response( # If not terminating, next_speaker must be provided thus will not be None next_speaker, # type: ignore[arg-type] cast(WorkflowContext[AgentExecutorRequest | GroupChatRequestMessage], ctx), - additional_instruction=_CONSECUTIVE_TURN_DEFAULT_INSTRUCTION if next_speaker == participant else None, + additional_instruction=( + _CONTINUATION_DEFAULT_INSTRUCTION if next_speaker == participant or not messages else None + ), ) self._increment_round() diff --git a/python/packages/orchestrations/tests/test_group_chat.py b/python/packages/orchestrations/tests/test_group_chat.py index 8436e97e454..c9193fbe940 100644 --- a/python/packages/orchestrations/tests/test_group_chat.py +++ b/python/packages/orchestrations/tests/test_group_chat.py @@ -33,7 +33,7 @@ from agent_framework_orchestrations import BaseGroupChatOrchestrator from agent_framework_orchestrations._group_chat import ( # pyright: ignore[reportPrivateUsage] - _CONSECUTIVE_TURN_DEFAULT_INSTRUCTION, + _CONTINUATION_DEFAULT_INSTRUCTION, ) @@ -1197,9 +1197,9 @@ async def test_group_chat_consecutive_selection_sends_non_empty_messages(stream: assert len(agent.received_messages) == 2, "Expected the participant to be invoked twice" assert all(received for received in agent.received_messages), "Participant was invoked with empty messages" - assert any( - _CONSECUTIVE_TURN_DEFAULT_INSTRUCTION in (message.text or "") for message in agent.received_messages[1] - ), "Second invocation should carry the continuation instruction" + assert any(_CONTINUATION_DEFAULT_INSTRUCTION in (message.text or "") for message in agent.received_messages[1]), ( + "Second invocation should carry the continuation instruction" + ) async def test_agent_orchestrator_consecutive_selection_sends_non_empty_messages() -> None: @@ -1216,9 +1216,9 @@ async def test_agent_orchestrator_consecutive_selection_sends_non_empty_messages assert len(agent.received_messages) == 2, "Expected the participant to be invoked twice" assert all(received for received in agent.received_messages), "Participant was invoked with empty messages" - assert any( - _CONSECUTIVE_TURN_DEFAULT_INSTRUCTION in (message.text or "") for message in agent.received_messages[1] - ), "Second invocation should carry the continuation instruction" + assert any(_CONTINUATION_DEFAULT_INSTRUCTION in (message.text or "") for message in agent.received_messages[1]), ( + "Second invocation should carry the continuation instruction" + ) async def test_group_chat_alternating_selection_has_no_continuation_instruction() -> None: @@ -1238,7 +1238,61 @@ async def test_group_chat_alternating_selection_has_no_continuation_instruction( for participant in (alpha, beta): for received in participant.received_messages: assert received, "Participant was invoked with empty messages" - assert all(_CONSECUTIVE_TURN_DEFAULT_INSTRUCTION not in (message.text or "") for message in received) + assert all(_CONTINUATION_DEFAULT_INSTRUCTION not in (message.text or "") for message in received) + + +class ToolOnlyStubAgent(BaseAgent): + """Stub agent whose response holds only a function call, so cleaning strips it entirely.""" + + def __init__(self, agent_name: str, **kwargs: Any) -> None: + super().__init__(name=agent_name, description=f"Tool-only stub agent {agent_name}", **kwargs) + + def run( # type: ignore[override] + self, + messages: str | Content | Message | Sequence[str | Content | Message] | None = None, + *, + stream: bool = False, + session: AgentSession | None = None, + **kwargs: Any, + ) -> Any: + return self._run_stream_impl() if stream else self._run_impl() + + def _contents(self) -> list[Content]: + return [Content.from_function_call(call_id=f"call-{self.name}", name="do_work", arguments={})] + + async def _run_impl(self) -> AgentResponse[Any]: + return AgentResponse(messages=[Message(role="assistant", contents=self._contents(), author_name=self.name)]) + + async def _run_stream_impl(self) -> AsyncIterable[AgentResponseUpdate]: + yield AgentResponseUpdate(contents=self._contents(), role="assistant", author_name=self.name) + + +async def test_group_chat_empty_cleaned_broadcast_sends_non_empty_messages() -> None: + """A broadcast cleaned down to nothing must not leave the next speaker with an empty cache. + + ``clean_conversation_for_handoff`` drops messages with no text content, so a tool-only + response broadcasts nothing and the next speaker is a *different* participant than the + one that just spoke. + """ + alpha = RecordingStubAgent("alpha", "reply from alpha") + beta = ToolOnlyStubAgent("beta") + + speakers = iter(["alpha", "beta", "alpha"]) + + workflow = GroupChatBuilder( + participants=[alpha, beta], + max_rounds=3, + selection_func=lambda state: next(speakers), + ).build() + + async for _ in workflow.run("kickoff", stream=True): + pass + + assert len(alpha.received_messages) == 2, "Expected alpha to be invoked twice" + assert all(received for received in alpha.received_messages), "Participant was invoked with empty messages" + assert any(_CONTINUATION_DEFAULT_INSTRUCTION in (message.text or "") for message in alpha.received_messages[1]), ( + "Second invocation should carry the continuation instruction" + ) # endregion