From 66ef1cfa840a4267a86bbc04a601d073877f594a Mon Sep 17 00:00:00 2001 From: Giles Odigwe Date: Thu, 6 Aug 2026 15:26:53 -0700 Subject: [PATCH 1/5] 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/5] 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/5] 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 From cc7c43b6698d69c686e2e263f971f86d9c39f359 Mon Sep 17 00:00:00 2001 From: Giles Odigwe Date: Mon, 10 Aug 2026 15:31:10 -0700 Subject: [PATCH 4/5] Only send the continuation instruction on an empty agent cache The previous revision also sent the instruction whenever the cleaned broadcast was empty. That is not the same question as "is the selected participant's cache empty": on the first round there is no prior turn to broadcast, so a tool-only participant speaking first received "Continue the conversation." in place of the user's task. Narrow the condition back to the re-selected speaker, which is exact: AgentExecutor clears its cache immediately after emitting a response, so the orchestrator receiving a response from a participant implies that participant's cache is now empty. Also restrict it to agent participants. Custom executors have no cache, and additional_instruction is a user-visible field on GroupChatRequestMessage, so sending it there would change observable behaviour without preventing any crash. Add regression tests for a tool-only participant speaking first and for a consecutively selected custom executor. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 149dd7ec-b2cc-4108-87bc-df347061932b --- .../_group_chat.py | 18 +++--- .../orchestrations/tests/test_group_chat.py | 60 +++++++++++++++---- 2 files changed, 59 insertions(+), 19 deletions(-) diff --git a/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py b/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py index 525629c7cf5..db36970bc53 100644 --- a/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py +++ b/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py @@ -69,11 +69,11 @@ logger = logging.getLogger(__name__) -# 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. +# Sent when the agent participant that just spoke is selected again. It is excluded from the +# broadcast (its own reply is already in its session) and AgentExecutor clears its cache after +# each run, so the request would otherwise carry no messages and some agents (for example +# A2AAgent) reject empty input. Custom executors are excluded: they have no such cache and +# receive full context in the request envelope. _CONTINUATION_DEFAULT_INSTRUCTION = "Continue the conversation." @@ -241,7 +241,9 @@ async def _handle_response( next_speaker, cast(WorkflowContext[AgentExecutorRequest | GroupChatRequestMessage], ctx), additional_instruction=( - _CONTINUATION_DEFAULT_INSTRUCTION if next_speaker == participant or not messages else None + _CONTINUATION_DEFAULT_INSTRUCTION + if next_speaker == participant and self._participant_registry.is_agent(participant) + else None ), ) self._increment_round() @@ -425,7 +427,9 @@ async def _handle_response( next_speaker, # type: ignore[arg-type] cast(WorkflowContext[AgentExecutorRequest | GroupChatRequestMessage], ctx), additional_instruction=( - _CONTINUATION_DEFAULT_INSTRUCTION if next_speaker == participant or not messages else None + _CONTINUATION_DEFAULT_INSTRUCTION + if next_speaker == participant and self._participant_registry.is_agent(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 c9193fbe940..72338e28e8e 100644 --- a/python/packages/orchestrations/tests/test_group_chat.py +++ b/python/packages/orchestrations/tests/test_group_chat.py @@ -15,15 +15,19 @@ ChatResponse, ChatResponseUpdate, Content, + Executor, Message, + WorkflowContext, WorkflowEvent, WorkflowRunState, + handler, ) from agent_framework._workflows._checkpoint import InMemoryCheckpointStorage from agent_framework._workflows._message_utils import normalize_messages_input from agent_framework.orchestrations import ( AgentRequestInfoResponse, GroupChatBuilder, + GroupChatRequestMessage, GroupChatState, MagenticContext, MagenticManagerBase, @@ -32,6 +36,7 @@ ) from agent_framework_orchestrations import BaseGroupChatOrchestrator +from agent_framework_orchestrations._base_group_chat_orchestrator import GroupChatResponseMessage from agent_framework_orchestrations._group_chat import ( # pyright: ignore[reportPrivateUsage] _CONTINUATION_DEFAULT_INSTRUCTION, ) @@ -1267,32 +1272,63 @@ 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. +async def test_group_chat_tool_only_first_speaker_keeps_initial_task() -> None: + """An empty broadcast must not displace context the next speaker already holds. - ``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. + ``clean_conversation_for_handoff`` drops messages with no text, so a tool-only first + response broadcasts nothing. The next speaker still has the initial task cached and must + receive it, rather than a continuation instruction that would replace it. """ alpha = RecordingStubAgent("alpha", "reply from alpha") beta = ToolOnlyStubAgent("beta") - speakers = iter(["alpha", "beta", "alpha"]) + speakers = iter(["beta", "alpha"]) workflow = GroupChatBuilder( participants=[alpha, beta], - max_rounds=3, + max_rounds=2, 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" - ) + assert len(alpha.received_messages) == 1, "Expected alpha to be invoked once" + received = alpha.received_messages[0] + assert any("kickoff" in (message.text or "") for message in received), "Initial task was dropped" + assert all(_CONTINUATION_DEFAULT_INSTRUCTION not in (message.text or "") for message in received) + + +class RecordingCustomExecutor(Executor): + """Custom executor participant that records the request envelopes it receives.""" + + def __init__(self, name: str) -> None: + super().__init__(name) + self.received_instructions: list[str | None] = [] + + @handler + async def _respond(self, message: GroupChatRequestMessage, ctx: WorkflowContext[GroupChatResponseMessage]) -> None: + self.received_instructions.append(message.additional_instruction) + await ctx.send_message( + GroupChatResponseMessage(message=Message(role="assistant", contents=["custom reply"], author_name=self.id)) + ) + + +async def test_group_chat_consecutive_custom_executor_gets_no_continuation_instruction() -> None: + """Custom executors have no agent cache, so re-selecting one must not add the instruction.""" + custom = RecordingCustomExecutor("custom") + + workflow = GroupChatBuilder( + participants=[custom], + max_rounds=2, + selection_func=lambda state: "custom", + ).build() + + async for _ in workflow.run("kickoff", stream=True): + pass + + assert len(custom.received_instructions) == 2, "Expected the participant to be invoked twice" + assert custom.received_instructions == [None, None] # endregion From 2a35005fccb69ea0a791050e5864932b602ddd09 Mon Sep 17 00:00:00 2001 From: Giles Odigwe Date: Tue, 11 Aug 2026 12:10:14 -0700 Subject: [PATCH 5/5] Apply the continuation fallback in AgentExecutor instead of the orchestrator The orchestrator cannot reliably tell whether a participant's cache is empty. Gating on "the participant just spoke" missed the non-consecutive case: if A speaks and its cache is cleared, B then returns tool-only content that cleaning strips to nothing, and A is selected again, A is invoked with no messages and A2AAgent raises the original ValueError. Move the decision to the component that owns the cache. AgentExecutorRequest gains an optional fallback_messages field, applied by AgentExecutor.run only when the cache would otherwise be empty, so it can never displace valid context. Group chat passes the continuation instruction unconditionally and drops its own condition entirely. Add the non-consecutive regression case and core tests covering both the applied and ignored paths. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 149dd7ec-b2cc-4108-87bc-df347061932b --- .../_workflows/_agent_executor.py | 9 ++++- .../tests/workflow/test_agent_executor.py | 36 +++++++++++++++++++ .../_base_group_chat_orchestrator.py | 12 ++++++- .../_group_chat.py | 25 +++++-------- .../orchestrations/tests/test_group_chat.py | 26 ++++++++++++++ 5 files changed, 89 insertions(+), 19 deletions(-) diff --git a/python/packages/core/agent_framework/_workflows/_agent_executor.py b/python/packages/core/agent_framework/_workflows/_agent_executor.py index b7787736fa5..737ddc69198 100644 --- a/python/packages/core/agent_framework/_workflows/_agent_executor.py +++ b/python/packages/core/agent_framework/_workflows/_agent_executor.py @@ -3,7 +3,7 @@ import logging import sys from collections.abc import Awaitable, Callable -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import Any, Literal, cast from typing_extensions import Never @@ -37,10 +37,15 @@ class AgentExecutorRequest: messages: A list of chat messages to be processed by the agent. should_respond: A flag indicating whether the agent should respond to the messages. If False, the messages will be saved to the executor's cache but not sent to the agent. + fallback_messages: Messages used only if the agent would otherwise run with an empty + cache. The cache is cleared after every response, so a caller that cannot know + whether earlier context survived can supply a fallback here instead of guessing. + Ignored whenever the cache holds anything, so valid context is never displaced. """ messages: list[Message] should_respond: bool = True + fallback_messages: list[Message] = field(default_factory=lambda: list[Message]()) @dataclass @@ -208,6 +213,8 @@ async def run( self._cache.extend(request.messages) if request.should_respond: + if not self._cache: + self._cache.extend(request.fallback_messages) await self._run_agent_and_emit(ctx) @handler diff --git a/python/packages/core/tests/workflow/test_agent_executor.py b/python/packages/core/tests/workflow/test_agent_executor.py index ccb1e9425bf..d4ff6d9ef94 100644 --- a/python/packages/core/tests/workflow/test_agent_executor.py +++ b/python/packages/core/tests/workflow/test_agent_executor.py @@ -7,6 +7,7 @@ from agent_framework import ( AgentExecutor, + AgentExecutorRequest, AgentResponse, AgentResponseUpdate, AgentRunInputs, @@ -867,3 +868,38 @@ async def test_agent_executor_request_info_uses_user_input_request_id() -> None: # endregion Tool approval emission + + +async def _run_request(agent: _MessageCapturingAgent, request: AgentExecutorRequest) -> None: + executor = AgentExecutor(agent, id="exec") + wf = WorkflowBuilder(start_executor=executor).build() + async for ev in wf.run(request, stream=True): + if ev.type == "status" and ev.state == WorkflowRunState.IDLE: + break + + +async def test_fallback_messages_used_when_cache_is_empty() -> None: + """A request carrying no messages must fall back rather than invoke the agent with nothing.""" + agent = _MessageCapturingAgent(id="a", name="A") + + await _run_request( + agent, + AgentExecutorRequest(messages=[], fallback_messages=[Message("user", ["carry on"])]), + ) + + assert [m.text for m in agent.last_messages] == ["carry on"] + + +async def test_fallback_messages_ignored_when_cache_has_content() -> None: + """Fallback messages must never displace real context.""" + agent = _MessageCapturingAgent(id="a", name="A") + + await _run_request( + agent, + AgentExecutorRequest( + messages=[Message("user", ["real task"])], + fallback_messages=[Message("user", ["carry on"])], + ), + ) + + assert [m.text for m in agent.last_messages] == ["real task"] diff --git a/python/packages/orchestrations/agent_framework_orchestrations/_base_group_chat_orchestrator.py b/python/packages/orchestrations/agent_framework_orchestrations/_base_group_chat_orchestrator.py index 16061b726b6..212ef9f0bf0 100644 --- a/python/packages/orchestrations/agent_framework_orchestrations/_base_group_chat_orchestrator.py +++ b/python/packages/orchestrations/agent_framework_orchestrations/_base_group_chat_orchestrator.py @@ -446,6 +446,7 @@ async def _send_request_to_participant( ctx: WorkflowContext[AgentExecutorRequest | GroupChatRequestMessage], *, additional_instruction: str | None = None, + fallback_instruction: str | None = None, metadata: dict[str, Any] | None = None, ) -> None: """Send a request to a participant. @@ -459,6 +460,9 @@ async def _send_request_to_participant( ctx: Workflow context for message routing additional_instruction: Optional additional instruction for the participant. This can be used to provide guidance to steer the participant's response. + fallback_instruction: Optional instruction applied only if the agent would + otherwise run with no messages at all. Ignored for custom executors, which + keep no message cache and always receive the full context envelope. metadata: Optional metadata dict Raises: @@ -469,7 +473,13 @@ async def _send_request_to_participant( messages: list[Message] = [] if additional_instruction: messages.append(Message(role="user", contents=[additional_instruction])) - request = AgentExecutorRequest(messages=messages, should_respond=True) + request = AgentExecutorRequest( + messages=messages, + should_respond=True, + fallback_messages=( + [Message(role="user", contents=[fallback_instruction])] if fallback_instruction else [] + ), + ) await ctx.send_message(request, target_id=target) await ctx.add_event( WorkflowEvent( diff --git a/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py b/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py index db36970bc53..76a5e526ce5 100644 --- a/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py +++ b/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py @@ -69,11 +69,11 @@ logger = logging.getLogger(__name__) -# Sent when the agent participant that just spoke is selected again. It is excluded from the -# broadcast (its own reply is already in its session) and AgentExecutor clears its cache after -# each run, so the request would otherwise carry no messages and some agents (for example -# A2AAgent) reject empty input. Custom executors are excluded: they have no such cache and -# receive full context in the request envelope. +# Applied only when the selected agent would otherwise run with no messages at all. This +# happens when it just spoke (it is excluded from the broadcast) or when cleaning stripped +# every broadcast message, since AgentExecutor clears its cache after each response. Some +# agents (for example A2AAgent) reject empty input. AgentExecutor drops this whenever real +# context is present, so it never displaces the conversation. _CONTINUATION_DEFAULT_INSTRUCTION = "Continue the conversation." @@ -240,11 +240,7 @@ async def _handle_response( await self._send_request_to_participant( next_speaker, cast(WorkflowContext[AgentExecutorRequest | GroupChatRequestMessage], ctx), - additional_instruction=( - _CONTINUATION_DEFAULT_INSTRUCTION - if next_speaker == participant and self._participant_registry.is_agent(participant) - else None - ), + fallback_instruction=_CONTINUATION_DEFAULT_INSTRUCTION, ) self._increment_round() @@ -421,16 +417,11 @@ 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 - next_speaker, # type: ignore[arg-type] + agent_orchestration_output.next_speaker, # type: ignore[arg-type] cast(WorkflowContext[AgentExecutorRequest | GroupChatRequestMessage], ctx), - additional_instruction=( - _CONTINUATION_DEFAULT_INSTRUCTION - if next_speaker == participant and self._participant_registry.is_agent(participant) - else None - ), + fallback_instruction=_CONTINUATION_DEFAULT_INSTRUCTION, ) self._increment_round() diff --git a/python/packages/orchestrations/tests/test_group_chat.py b/python/packages/orchestrations/tests/test_group_chat.py index 72338e28e8e..9034e7e7c15 100644 --- a/python/packages/orchestrations/tests/test_group_chat.py +++ b/python/packages/orchestrations/tests/test_group_chat.py @@ -1299,6 +1299,32 @@ async def test_group_chat_tool_only_first_speaker_keeps_initial_task() -> None: assert all(_CONTINUATION_DEFAULT_INSTRUCTION not in (message.text or "") for message in received) +async def test_group_chat_empty_broadcast_after_earlier_speaker_sends_non_empty_messages() -> None: + """A non-consecutive speaker whose cache is empty must still receive messages. + + ``alpha`` speaks and its cache is cleared. ``beta`` then returns tool-only content, which + ``clean_conversation_for_handoff`` strips to nothing, so the broadcast back to ``alpha`` + adds nothing. Re-selecting ``alpha`` would otherwise invoke it with no messages at all. + """ + 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]) + + class RecordingCustomExecutor(Executor): """Custom executor participant that records the request envelopes it receives."""