Python: Fix group chat invoking a re-selected participant with no messages

python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py

@@ -69,6 +69,13 @@
 
 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.
+_CONTINUATION_DEFAULT_INSTRUCTION = "Continue the conversation."
+
 
 @dataclass(frozen=True)
 class GroupChatState:
@@ -233,6 +240,11 @@ 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
+            ),
         )
         self._increment_round()
 
@@ -409,10 +421,16 @@ 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=(
+                _CONTINUATION_DEFAULT_INSTRUCTION
+                if next_speaker == participant and self._participant_registry.is_agent(participant)
+                else None
+            ),
         )
         self._increment_round()
 

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
 
@@ -14,14 +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,
@@ -30,6 +36,10 @@
 )
 
 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,
+)
 
 
 class StubAgent(BaseAgent):
@@ -1098,3 +1108,227 @@ 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 = 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).")
+        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(_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:
+    """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(_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:
+    """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(_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_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, 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(["beta", "alpha"])
+
+    workflow = GroupChatBuilder(
+        participants=[alpha, beta],
+        max_rounds=2,
+        selection_func=lambda state: next(speakers),
+    ).build()
+
+    async for _ in workflow.run("kickoff", stream=True):
+        pass
+
+    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