Repository: microsoft/autogen
Pull request: Fix output task messages 6150
## Why are these changes needed?

The existing run_stream methods used fragile count-based logic (count <= len(task)) to skip task messages during streaming. This approach was brittle and broke when team structure changed or task composition varied, particularly affecting SocietyOfMindAgent's ability to properly encapsulate inner team messages.

This PR adds an output_task_messages parameter to run_stream methods to provide explicit control over task message inclusion in streams, replacing the fragile count-based logic with robust message filtering.

## Related issue number

Closes #6150

## Checks

- [ ] I've included any doc changes needed for <https://microsoft.github.io/autogen/>. See <https://github.com/microsoft/autogen/blob/main/CONTRIBUTING.md> to build and test documentation locally.
- [x] I've added tests corresponding to the changes introduced in this PR.
- [x] I've made sure all auto checks have passed.

Problem statement:
Adding a `output_task_messages` to `run_stream` methods to avoid messages in `task` are included.
### Confirmation

- [x] I confirm that I am a maintainer and so can use this template. If I am not, I understand this issue will be closed and I will be asked to use a different template.

### Issue body

Help `SocietyOfMindAgent` to avoid fragile counting-based skipping of task messages. Also, it makes it easier to nest agent teams within teams.

Hints from issue discussion:

Commit message: refactor: added output_task_messages flag in the BaseGroupChatManager

Patch to review:
diff --git a/python/packages/autogen-agentchat/src/autogen_agentchat/agents/_base_chat_agent.py b/python/packages/autogen-agentchat/src/autogen_agentchat/agents/_base_chat_agent.py
index 6f9c067428ea..12b15944879a 100644
--- a/python/packages/autogen-agentchat/src/autogen_agentchat/agents/_base_chat_agent.py
+++ b/python/packages/autogen-agentchat/src/autogen_agentchat/agents/_base_chat_agent.py
@@ -153,9 +153,16 @@ async def run_stream(
         *,
         task: str | BaseChatMessage | Sequence[BaseChatMessage] | None = None,
         cancellation_token: CancellationToken | None = None,
+        output_task_messages: bool = True,
     ) -> AsyncGenerator[BaseAgentEvent | BaseChatMessage | TaskResult, None]:
         """Run the agent with the given task and return a stream of messages
-        and the final task result as the last item in the stream."""
+        and the final task result as the last item in the stream.
+
+        Args:
+            task: The task to run. Can be a string, a single message, or a sequence of messages.
+            cancellation_token: The cancellation token to kill the task immediately.
+            output_task_messages: Whether to include task messages in the output stream. Defaults to True for backward compatibility.
+        """
         with trace_invoke_agent_span(
             agent_name=self.name,
             agent_description=self.description,
@@ -169,20 +176,23 @@ async def run_stream(
             elif isinstance(task, str):
                 text_msg = TextMessage(content=task, source="user")
                 input_messages.append(text_msg)
-                output_messages.append(text_msg)
-                yield text_msg
+                if output_task_messages:
+                    output_messages.append(text_msg)
+                    yield text_msg
             elif isinstance(task, BaseChatMessage):
                 input_messages.append(task)
-                output_messages.append(task)
-                yield task
+                if output_task_messages:
+                    output_messages.append(task)
+                    yield task
             else:
                 if not task:
                     raise ValueError("Task list cannot be empty.")
                 for msg in task:
                     if isinstance(msg, BaseChatMessage):
                         input_messages.append(msg)
-                        output_messages.append(msg)
-                        yield msg
+                        if output_task_messages:
+                            output_messages.append(msg)
+                            yield msg
                     else:
                         raise ValueError(f"Invalid message type in sequence: {type(msg)}")
             async for message in self.on_messages_stream(input_messages, cancellation_token):
diff --git a/python/packages/autogen-agentchat/src/autogen_agentchat/agents/_society_of_mind_agent.py b/python/packages/autogen-agentchat/src/autogen_agentchat/agents/_society_of_mind_agent.py
index 9ed9606362c6..31f26db65ce1 100644
--- a/python/packages/autogen-agentchat/src/autogen_agentchat/agents/_society_of_mind_agent.py
+++ b/python/packages/autogen-agentchat/src/autogen_agentchat/agents/_society_of_mind_agent.py
@@ -175,7 +175,6 @@ async def on_messages_stream(
         result: TaskResult | None = None
         inner_messages: List[BaseAgentEvent | BaseChatMessage] = []
         model_context = self._model_context
-        count = 0
 
         prev_content = await model_context.get_messages()
         if len(prev_content) > 0:
@@ -192,14 +191,13 @@ async def on_messages_stream(
         else:
             task = task_messages
 
-        async for inner_msg in self._team.run_stream(task=task, cancellation_token=cancellation_token):
+        # Use the new output_task_messages parameter to avoid fragile count-based logic
+        async for inner_msg in self._team.run_stream(
+            task=task, cancellation_token=cancellation_token, output_task_messages=False
+        ):
             if isinstance(inner_msg, TaskResult):
                 result = inner_msg
             else:
-                count += 1
-                if count <= len(task_messages):
-                    # Skip the task messages.
-                    continue
                 yield inner_msg
                 if isinstance(inner_msg, ModelClientStreamingChunkEvent):
                     # Skip the model client streaming chunk events.
diff --git a/python/packages/autogen-agentchat/src/autogen_agentchat/base/_task.py b/python/packages/autogen-agentchat/src/autogen_agentchat/base/_task.py
index 747535c90a4f..ddd2e5ff57fb 100644
--- a/python/packages/autogen-agentchat/src/autogen_agentchat/base/_task.py
+++ b/python/packages/autogen-agentchat/src/autogen_agentchat/base/_task.py
@@ -39,6 +39,7 @@ def run_stream(
         *,
         task: str | BaseChatMessage | Sequence[BaseChatMessage] | None = None,
         cancellation_token: CancellationToken | None = None,
+        output_task_messages: bool = True,
     ) -> AsyncGenerator[BaseAgentEvent | BaseChatMessage | TaskResult, None]:
         """Run the task and produces a stream of messages and the final result
         :class:`TaskResult` as the last item in the stream.
@@ -47,5 +48,11 @@ def run_stream(
 
         The runner is stateful and a subsequent call to this method will continue
         from where the previous call left off. If the task is not specified,
-        the runner will continue with the current task."""
+        the runner will continue with the current task.
+
+        Args:
+            task: The task to run. Can be a string, a single message, or a sequence of messages.
+            cancellation_token: The cancellation token to kill the task immediately.
+            output_task_messages: Whether to include task messages in the output stream. Defaults to True for backward compatibility.
+        """
         ...
diff --git a/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_base_group_chat.py b/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_base_group_chat.py
index 0d81c391389c..4340af2ac259 100644
--- a/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_base_group_chat.py
+++ b/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_base_group_chat.py
@@ -66,6 +66,7 @@ def __init__(
         self._termination_condition = termination_condition
         self._max_turns = max_turns
         self._message_factory = MessageFactory()
+        self._output_task_messages = True  # Default value, will be updated in run_stream
         if custom_message_types is not None:
             for message_type in custom_message_types:
                 self._message_factory.register(message_type)
@@ -141,6 +142,7 @@ def _create_group_chat_manager_factory(
         termination_condition: TerminationCondition | None,
         max_turns: int | None,
         message_factory: MessageFactory,
+        output_task_messages: bool,
     ) -> Callable[[], SequentialRoutedAgent]: ...
 
     def _create_participant_factory(
@@ -192,6 +194,7 @@ async def _init(self, runtime: AgentRuntime) -> None:
                 termination_condition=self._termination_condition,
                 max_turns=self._max_turns,
                 message_factory=self._message_factory,
+                output_task_messages=self._output_task_messages,
             ),
         )
         # Add subscriptions for the group chat manager.
@@ -319,6 +322,7 @@ async def run_stream(
         *,
         task: str | BaseChatMessage | Sequence[BaseChatMessage] | None = None,
         cancellation_token: CancellationToken | None = None,
+        output_task_messages: bool = True,
     ) -> AsyncGenerator[BaseAgentEvent | BaseChatMessage | TaskResult, None]:
         """Run the team and produces a stream of messages and the final result
         of the type :class:`~autogen_agentchat.base.TaskResult` as the last item in the stream. Once the
@@ -336,6 +340,7 @@ async def run_stream(
                 Setting the cancellation token potentially put the team in an inconsistent state,
                 and it may not reset the termination condition.
                 To gracefully stop the team, use :class:`~autogen_agentchat.conditions.ExternalTermination` instead.
+            output_task_messages (bool): Whether to include task messages in the output stream. Defaults to True for backward compatibility.
 
         Returns:
             stream: an :class:`~collections.abc.AsyncGenerator` that yields :class:`~autogen_agentchat.messages.BaseAgentEvent`, :class:`~autogen_agentchat.messages.BaseChatMessage`, and the final result :class:`~autogen_agentchat.base.TaskResult` as the last item in the stream.
@@ -416,6 +421,8 @@ async def main() -> None:
             asyncio.run(main())
 
         """
+        # Update the output_task_messages field with the provided value
+        self._output_task_messages = output_task_messages
 
         # Create the messages list if the task is a string or a chat message.
         messages: List[BaseChatMessage] | None = None
@@ -504,7 +511,18 @@ async def stop_runtime() -> None:
             # Collect the output messages in order.
             output_messages: List[BaseAgentEvent | BaseChatMessage] = []
             stop_reason: str | None = None
-            # Yield the messsages until the queue is empty.
+            task_messages_list: List[BaseChatMessage] = []
+
+            # Add task messages to output_messages and stream only if output_task_messages is True
+            # Also track the actual task message objects for later filtering
+            if messages is not None:
+                for msg in messages:
+                    task_messages_list.append(msg)  # Track actual message objects
+                    if output_task_messages:
+                        output_messages.append(msg)  # Only add to output_messages if output_task_messages is True
+                        yield msg  # Only yield to stream if output_task_messages is True
+
+            # Yield the messages until the queue is empty.
             while True:
                 message_future = asyncio.ensure_future(self._output_message_queue.get())
                 if cancellation_token is not None:
@@ -518,6 +536,11 @@ async def stop_runtime() -> None:
                         raise RuntimeError(str(message.error))
                     stop_reason = message.message.content
                     break
+
+                # Skip initial task messages (either yielded initially or queued by the manager)
+                if message in task_messages_list:
+                    continue  # Skip task messages
+
                 yield message
                 if isinstance(message, ModelClientStreamingChunkEvent):
                     # Skip the model client streaming chunk events.
diff --git a/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_base_group_chat_manager.py b/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_base_group_chat_manager.py
index 6c008bef6667..d2e2683b0b06 100644
--- a/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_base_group_chat_manager.py
+++ b/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_base_group_chat_manager.py
@@ -46,6 +46,7 @@ def __init__(
         max_turns: int | None,
         message_factory: MessageFactory,
         emit_team_events: bool = False,
+        output_task_messages: bool = True,
     ):
         super().__init__(
             description="Group chat manager",
@@ -79,6 +80,7 @@ def __init__(
         self._current_turn = 0
         self._message_factory = message_factory
         self._emit_team_events = emit_team_events
+        self._output_task_messages = output_task_messages
         self._active_speakers: List[str] = []
 
     @rpc
@@ -105,8 +107,11 @@ async def handle_start(self, message: GroupChatStart, ctx: MessageContext) -> No
                 GroupChatStart(messages=message.messages),
                 topic_id=DefaultTopicId(type=self._output_topic_type),
             )
-            for msg in message.messages:
-                await self._output_message_queue.put(msg)
+
+            # Only put messages in output queue if they are not task messages or output_task_messages is True
+            if self._output_task_messages:
+                for msg in message.messages:
+                    await self._output_message_queue.put(msg)
 
             # Relay all messages at once to participants
             await self.publish_message(
diff --git a/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_graph/_digraph_group_chat.py b/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_graph/_digraph_group_chat.py
index aaca418df92b..a49842027627 100644
--- a/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_graph/_digraph_group_chat.py
+++ b/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_graph/_digraph_group_chat.py
@@ -326,6 +326,7 @@ def __init__(
         max_turns: int | None,
         message_factory: MessageFactory,
         graph: DiGraph,
+        output_task_messages: bool = True,
     ) -> None:
         """Initialize the graph-based execution manager."""
         super().__init__(
@@ -339,6 +340,7 @@ def __init__(
             termination_condition=termination_condition,
             max_turns=max_turns,
             message_factory=message_factory,
+            output_task_messages=output_task_messages,
         )
         graph.graph_validate()
         if graph.get_has_cycles() and self._termination_condition is None and self._max_turns is None:
@@ -805,6 +807,7 @@ def _create_group_chat_manager_factory(
         termination_condition: TerminationCondition | None,
         max_turns: int | None,
         message_factory: MessageFactory,
+        output_task_messages: bool,
     ) -> Callable[[], GraphFlowManager]:
         """Creates the factory method for initializing the DiGraph-based chat manager."""
 
@@ -821,6 +824,7 @@ def _factory() -> GraphFlowManager:
                 max_turns=max_turns,
                 message_factory=message_factory,
                 graph=self._graph,
+                output_task_messages=output_task_messages,
             )
 
         return _factory
diff --git a/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_magentic_one/_magentic_one_group_chat.py b/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_magentic_one/_magentic_one_group_chat.py
index 3b637d459573..eeced8d18807 100644
--- a/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_magentic_one/_magentic_one_group_chat.py
+++ b/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_magentic_one/_magentic_one_group_chat.py
@@ -141,6 +141,7 @@ def _create_group_chat_manager_factory(
         termination_condition: TerminationCondition | None,
         max_turns: int | None,
         message_factory: MessageFactory,
+        output_task_messages: bool,
     ) -> Callable[[], MagenticOneOrchestrator]:
         return lambda: MagenticOneOrchestrator(
             name,
@@ -157,6 +158,7 @@ def _create_group_chat_manager_factory(
             output_message_queue,
             termination_condition,
             self._emit_team_events,
+            output_task_messages,
         )
 
     def _to_config(self) -> MagenticOneGroupChatConfig:
diff --git a/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_magentic_one/_magentic_one_orchestrator.py b/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_magentic_one/_magentic_one_orchestrator.py
index 2f1d2fd673f6..4fec4711e84a 100644
--- a/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_magentic_one/_magentic_one_orchestrator.py
+++ b/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_magentic_one/_magentic_one_orchestrator.py
@@ -73,6 +73,7 @@ def __init__(
         output_message_queue: asyncio.Queue[BaseAgentEvent | BaseChatMessage | GroupChatTermination],
         termination_condition: TerminationCondition | None,
         emit_team_events: bool,
+        output_task_messages: bool = True,
     ):
         super().__init__(
             name,
@@ -86,6 +87,7 @@ def __init__(
             max_turns,
             message_factory,
             emit_team_events=emit_team_events,
+            output_task_messages=output_task_messages,
         )
         self._model_client = model_client
         self._max_stalls = max_stalls
diff --git a/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_round_robin_group_chat.py b/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_round_robin_group_chat.py
index d6b43afb283f..70367a90b553 100644
--- a/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_round_robin_group_chat.py
+++ b/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_round_robin_group_chat.py
@@ -29,6 +29,7 @@ def __init__(
         max_turns: int | None,
         message_factory: MessageFactory,
         emit_team_events: bool,
+        output_task_messages: bool = True,
     ) -> None:
         super().__init__(
             name,
@@ -42,6 +43,7 @@ def __init__(
             max_turns,
             message_factory,
             emit_team_events,
+            output_task_messages,
         )
         self._next_speaker_index = 0
 
@@ -204,6 +206,7 @@ def _create_group_chat_manager_factory(
         termination_condition: TerminationCondition | None,
         max_turns: int | None,
         message_factory: MessageFactory,
+        output_task_messages: bool,
     ) -> Callable[[], RoundRobinGroupChatManager]:
         def _factory() -> RoundRobinGroupChatManager:
             return RoundRobinGroupChatManager(
@@ -218,6 +221,7 @@ def _factory() -> RoundRobinGroupChatManager:
                 max_turns,
                 message_factory,
                 self._emit_team_events,
+                output_task_messages,
             )
 
         return _factory
diff --git a/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_selector_group_chat.py b/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_selector_group_chat.py
index eaa3c320dab3..e5db73eca702 100644
--- a/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_selector_group_chat.py
+++ b/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_selector_group_chat.py
@@ -73,6 +73,7 @@ def __init__(
         emit_team_events: bool,
         model_context: ChatCompletionContext | None,
         model_client_streaming: bool = False,
+        output_task_messages: bool = True,
     ) -> None:
         super().__init__(
             name,
@@ -86,6 +87,7 @@ def __init__(
             max_turns,
             message_factory,
             emit_team_events,
+            output_task_messages,
         )
         self._model_client = model_client
         self._selector_prompt = selector_prompt
@@ -633,6 +635,7 @@ def _create_group_chat_manager_factory(
         termination_condition: TerminationCondition | None,
         max_turns: int | None,
         message_factory: MessageFactory,
+        output_task_messages: bool,
     ) -> Callable[[], BaseGroupChatManager]:
         return lambda: SelectorGroupChatManager(
             name,
@@ -654,6 +657,7 @@ def _create_group_chat_manager_factory(
             self._emit_team_events,
             self._model_context,
             self._model_client_streaming,
+            output_task_messages,
         )
 
     def _to_config(self) -> SelectorGroupChatConfig:
diff --git a/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_swarm_group_chat.py b/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_swarm_group_chat.py
index 344994015488..f0c973ab53b6 100644
--- a/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_swarm_group_chat.py
+++ b/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_swarm_group_chat.py
@@ -28,6 +28,7 @@ def __init__(
         max_turns: int | None,
         message_factory: MessageFactory,
         emit_team_events: bool,
+        output_task_messages: bool = True,
     ) -> None:
         super().__init__(
             name,
@@ -41,6 +42,7 @@ def __init__(
             max_turns,
             message_factory,
             emit_team_events,
+            output_task_messages,
         )
         self._current_speaker = self._participant_names[0]
 
@@ -254,6 +256,7 @@ def _create_group_chat_manager_factory(
         termination_condition: TerminationCondition | None,
         max_turns: int | None,
         message_factory: MessageFactory,
+        output_task_messages: bool,
     ) -> Callable[[], SwarmGroupChatManager]:
         def _factory() -> SwarmGroupChatManager:
             return SwarmGroupChatManager(
@@ -268,6 +271,7 @@ def _factory() -> SwarmGroupChatManager:
                 max_turns,
                 message_factory,
                 self._emit_team_events,
+                output_task_messages,
             )
 
         return _factory
diff --git a/python/packages/autogen-agentchat/tests/test_group_chat.py b/python/packages/autogen-agentchat/tests/test_group_chat.py
index 9e15eb789136..d3137ca58135 100644
--- a/python/packages/autogen-agentchat/tests/test_group_chat.py
+++ b/python/packages/autogen-agentchat/tests/test_group_chat.py
@@ -261,23 +261,28 @@ async def test_round_robin_group_chat(runtime: AgentRuntime | None) -> None:
 
         assert result.stop_reason is not None and result.stop_reason == "Text 'TERMINATE' mentioned"
 
-        # Test streaming.
+        # Test streaming with default output_task_messages=True.
         model_client.reset()
-        index = 0
         await team.reset()
+        streamed_messages: List[BaseAgentEvent | BaseChatMessage] = []
+        final_stream_result: TaskResult | None = None
         async for message in team.run_stream(
             task="Write a program that prints 'Hello, world!'",
         ):
             if isinstance(message, TaskResult):
-                assert compare_task_results(message, result)
+                final_stream_result = message
             else:
-                assert compare_messages(message, result.messages[index])
-            index += 1
+                streamed_messages.append(message)
+        assert final_stream_result is not None
+        assert compare_task_results(final_stream_result, result)
+        # Verify streamed messages match the complete result.messages
+        assert len(streamed_messages) == len(result.messages)
+        for streamed_msg, expected_msg in zip(streamed_messages, result.messages, strict=False):
+            assert compare_messages(streamed_msg, expected_msg)
 
         # Test message input.
         # Text message.
         model_client.reset()
-        index = 0
         await team.reset()
         result_2 = await team.run(
             task=TextMessage(content="Write a program that prints 'Hello, world!'", source="user")
@@ -286,7 +291,6 @@ async def test_round_robin_group_chat(runtime: AgentRuntime | None) -> None:
 
         # Test multi-modal message.
         model_client.reset()
-        index = 0
         await team.reset()
         task = MultiModalMessage(content=["Write a program that prints 'Hello, world!'"], source="user")
         result_2 = await team.run(task=task)
@@ -333,18 +337,36 @@ async def test_round_robin_group_chat_with_team_event(runtime: AgentRuntime | No
         assert isinstance(result.messages[5], SelectSpeakerEvent)
         assert isinstance(result.messages[6], TextMessage)
 
-        # Test streaming.
+        # Test streaming with default output_task_messages=True.
         model_client.reset()
-        index = 0
         await team.reset()
+        streamed_messages: List[BaseAgentEvent | BaseChatMessage] = []
+        final_stream_result: TaskResult | None = None
         async for message in team.run_stream(
             task="Write a program that prints 'Hello, world!'",
         ):
             if isinstance(message, TaskResult):
-                assert compare_task_results(message, result)
+                final_stream_result = message
             else:
-                assert compare_messages(message, result.messages[index])
-            index += 1
+                streamed_messages.append(message)
+        assert final_stream_result is not None
+        assert compare_task_results(final_stream_result, result)
+        # Verify streamed messages match the complete result.messages
+        assert len(streamed_messages) == len(result.messages)
+        for streamed_msg, expected_msg in zip(streamed_messages, result.messages, strict=False):
+            assert compare_messages(streamed_msg, expected_msg)
+
+        # Test multi-modal message.
+        model_client.reset()
+        await team.reset()
+        task = MultiModalMessage(content=["Write a program that prints 'Hello, world!'"], source="user")
+        result_2 = await team.run(task=task)
+        assert isinstance(result.messages[0], TextMessage)
+        assert isinstance(result_2.messages[0], MultiModalMessage)
+        assert result.messages[0].content == task.content[0]
+        assert len(result.messages[1:]) == len(result_2.messages[1:])
+        for i in range(1, len(result.messages)):
+            assert compare_messages(result.messages[i], result_2.messages[i])
 
 
 @pytest.mark.asyncio
@@ -492,7 +514,7 @@ async def test_round_robin_group_chat_with_tools(runtime: AgentRuntime | None) -
     # Test streaming.
     await tool_use_agent._model_context.clear()  # pyright: ignore
     model_client.reset()
-    index = 0
+    result_index = 0  # Include task message in result since output_task_messages=True by default
     await team.reset()
     async for message in team.run_stream(
         task="Write a program that prints 'Hello, world!'",
@@ -500,13 +522,12 @@ async def test_round_robin_group_chat_with_tools(runtime: AgentRuntime | None) -
         if isinstance(message, TaskResult):
             assert compare_task_results(message, result)
         else:
-            assert compare_messages(message, result.messages[index])
-        index += 1
+            assert compare_messages(message, result.messages[result_index])
+            result_index += 1
 
     # Test Console.
     await tool_use_agent._model_context.clear()  # pyright: ignore
     model_client.reset()
-    index = 0
     await team.reset()
     result2 = await Console(team.run_stream(task="Write a program that prints 'Hello, world!'"))
     assert compare_task_results(result2, result)
@@ -681,7 +702,7 @@ async def test_selector_group_chat(runtime: AgentRuntime | None) -> None:
     # Test streaming.
     model_client.reset()
     agent1._count = 0  # pyright: ignore
-    index = 0
+    result_index = 0  # Include task message in result since output_task_messages=True by default
     await team.reset()
     async for message in team.run_stream(
         task="Write a program that prints 'Hello, world!'",
@@ -689,13 +710,12 @@ async def test_selector_group_chat(runtime: AgentRuntime | None) -> None:
         if isinstance(message, TaskResult):
             assert compare_task_results(message, result)
         else:
-            assert compare_messages(message, result.messages[index])
-        index += 1
+            assert compare_messages(message, result.messages[result_index])
+            result_index += 1
 
     # Test Console.
     model_client.reset()
     agent1._count = 0  # pyright: ignore
-    index = 0
     await team.reset()
     result2 = await Console(team.run_stream(task="Write a program that prints 'Hello, world!'"))
     assert compare_task_results(result2, result)
@@ -802,7 +822,7 @@ async def test_selector_group_chat_with_team_event(runtime: AgentRuntime | None)
     # Test streaming.
     model_client.reset()
     agent1._count = 0  # pyright: ignore
-    index = 0
+    result_index = 0  # Include task message in result since output_task_messages=True by default
     await team.reset()
     async for message in team.run_stream(
         task="Write a program that prints 'Hello, world!'",
@@ -810,8 +830,8 @@ async def test_selector_group_chat_with_team_event(runtime: AgentRuntime | None)
         if isinstance(message, TaskResult):
             assert compare_task_results(message, result)
         else:
-            assert compare_messages(message, result.messages[index])
-        index += 1
+            assert compare_messages(message, result.messages[result_index])
+            result_index += 1
 
 
 @pytest.mark.asyncio
@@ -908,19 +928,18 @@ async def test_selector_group_chat_two_speakers(runtime: AgentRuntime | None) ->
     # Test streaming.
     model_client.reset()
     agent1._count = 0  # pyright: ignore
-    index = 0
+    result_index = 0  # Include task message in result since output_task_messages=True by default
     await team.reset()
     async for message in team.run_stream(task="Write a program that prints 'Hello, world!'"):
         if isinstance(message, TaskResult):
             assert compare_task_results(message, result)
         else:
-            assert compare_messages(message, result.messages[index])
-        index += 1
+            assert compare_messages(message, result.messages[result_index])
+            result_index += 1
 
     # Test Console.
     model_client.reset()
     agent1._count = 0  # pyright: ignore
-    index = 0
     await team.reset()
     result2 = await Console(team.run_stream(task="Write a program that prints 'Hello, world!'"))
     assert compare_task_results(result2, result)
@@ -956,18 +975,17 @@ async def test_selector_group_chat_two_speakers_allow_repeated(runtime: AgentRun
 
     # Test streaming.
     model_client.reset()
-    index = 0
+    result_index = 0  # Include task message in result since output_task_messages=True by default
     await team.reset()
     async for message in team.run_stream(task="Write a program that prints 'Hello, world!'"):
         if isinstance(message, TaskResult):
             assert compare_task_results(message, result)
         else:
-            assert compare_messages(message, result.messages[index])
-        index += 1
+            assert compare_messages(message, result.messages[result_index])
+            result_index += 1
 
     # Test Console.
     model_client.reset()
-    index = 0
     await team.reset()
     result2 = await Console(team.run_stream(task="Write a program that prints 'Hello, world!'"))
     assert compare_task_results(result2, result)
@@ -1171,15 +1189,15 @@ async def test_swarm_handoff(runtime: AgentRuntime | None) -> None:
     )
 
     # Test streaming.
-    index = 0
+    result_index = 0  # Include task message in result since output_task_messages=True by default
     await team.reset()
     stream = team.run_stream(task="task")
     async for message in stream:
         if isinstance(message, TaskResult):
             assert compare_task_results(message, result)
         else:
-            assert compare_messages(message, result.messages[index])
-        index += 1
+            assert compare_messages(message, result.messages[result_index])
+            result_index += 1
 
     # Test save and load.
     state = await team.save_state()
@@ -1245,15 +1263,15 @@ async def test_swarm_handoff_with_team_events(runtime: AgentRuntime | None) -> N
     )
 
     # Test streaming.
-    index = 0
+    result_index = 0  # Include task message in result since output_task_messages=True by default
     await team.reset()
     stream = team.run_stream(task="task")
     async for message in stream:
         if isinstance(message, TaskResult):
             assert compare_task_results(message, result)
         else:
-            assert compare_messages(message, result.messages[index])
-        index += 1
+            assert compare_messages(message, result.messages[result_index])
+            result_index += 1
 
 
 @pytest.mark.asyncio
@@ -1363,20 +1381,19 @@ async def test_swarm_handoff_using_tool_calls(runtime: AgentRuntime | None) -> N
     # Test streaming.
     await agent1._model_context.clear()  # pyright: ignore
     model_client.reset()
-    index = 0
+    result_index = 0  # Include task message in result since output_task_messages=True by default
     await team.reset()
     stream = team.run_stream(task="task")
     async for message in stream:
         if isinstance(message, TaskResult):
             assert compare_task_results(message, result)
         else:
-            assert compare_messages(message, result.messages[index])
-        index += 1
+            assert compare_messages(message, result.messages[result_index])
+            result_index += 1
 
     # Test Console
     await agent1._model_context.clear()  # pyright: ignore
     model_client.reset()
-    index = 0
     await team.reset()
     result2 = await Console(team.run_stream(task="task"))
     assert compare_task_results(result2, result)
@@ -1600,13 +1617,13 @@ async def test_round_robin_group_chat_with_message_list(runtime: AgentRuntime |
 
     # Test with streaming
     await team.reset()
-    index = 0
+    result_index = 0  # Include the 3 task messages in result since output_task_messages=True by default
     async for message in team.run_stream(task=messages):
         if isinstance(message, TaskResult):
             assert compare_task_results(message, result)
         else:
-            assert compare_messages(message, result.messages[index])
-            index += 1
+            assert compare_messages(message, result.messages[result_index])
+            result_index += 1
 
     # Test with invalid message list
     with pytest.raises(ValueError, match="All messages in task list must be valid BaseChatMessage types"):
@@ -1804,17 +1821,28 @@ async def test_selector_group_chat_streaming(runtime: AgentRuntime | None) -> No
     # Test streaming
     await team.reset()
     model_client.reset()
-    index = 0
-    streaming: List[str] = []
-    async for message in team.run_stream(task="Write a program that prints 'Hello, world!'"):
+    result_index = 0  # Include task message in result since output_task_messages=True by default
+    streamed_chunks: List[str] = []
+    final_result: TaskResult | None = None
+    async for message in team.run_stream(
+        task="Write a program that prints 'Hello, world!'",
+    ):
         if isinstance(message, TaskResult):
+            final_result = message
             assert compare_task_results(message, result)
         elif isinstance(message, ModelClientStreamingChunkEvent):
-            streaming.append(message.content)
+            streamed_chunks.append(message.content)
         else:
-            if streaming:
+            if streamed_chunks:
                 assert isinstance(message, SelectorEvent)
-                assert message.content == "".join([chunk for chunk in streaming])
-                streaming = []
-            assert compare_messages(message, result.messages[index])
-            index += 1
+                assert message.content == "".join(streamed_chunks)
+                streamed_chunks = []
+            assert compare_messages(message, result.messages[result_index])
+            result_index += 1
+
+    # Verify we got the expected messages without relying on fragile ordering
+    assert final_result is not None
+    assert len(streamed_chunks) == 0  # All chunks should have been processed
+
+    # Content-based verification instead of index-based
+    # Note: The streaming test verifies the streaming behavior, not the final result content
diff --git a/python/packages/autogen-agentchat/tests/test_society_of_mind_agent.py b/python/packages/autogen-agentchat/tests/test_society_of_mind_agent.py
index 5bc9a010cbb9..875fa06a9335 100644
--- a/python/packages/autogen-agentchat/tests/test_society_of_mind_agent.py
+++ b/python/packages/autogen-agentchat/tests/test_society_of_mind_agent.py
@@ -1,10 +1,12 @@
 from types import MethodType
-from typing import Any, AsyncGenerator, Sequence
+from typing import Any, AsyncGenerator, List, Sequence
 
 import pytest
 import pytest_asyncio
 from autogen_agentchat.agents import AssistantAgent, SocietyOfMindAgent
-from autogen_agentchat.conditions import MaxMessageTermination
+from autogen_agentchat.base import TaskResult
+from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
+from autogen_agentchat.messages import BaseAgentEvent, BaseChatMessage, TextMessage
 from autogen_agentchat.teams import RoundRobinGroupChat
 from autogen_core import AgentRuntime, SingleThreadedAgentRuntime
 from autogen_core.models import CreateResult, LLMMessage, SystemMessage
@@ -59,6 +61,82 @@ async def test_society_of_mind_agent(runtime: AgentRuntime | None) -> None:
     assert loaded_soc_agent.name == "society_of_mind"
 
 
+@pytest.mark.asyncio
+async def test_society_of_mind_agent_output_task_messages_parameter(runtime: AgentRuntime | None) -> None:
+    """Test that output_task_messages parameter controls whether task messages are included in the stream."""
+    model_client = ReplayChatCompletionClient(
+        ["1", "2", "3", "4", "5", "6", "7", "8", "9", "10"],
+    )
+    agent1 = AssistantAgent("assistant1", model_client=model_client, system_message="You are a helpful assistant.")
+    agent2 = AssistantAgent("assistant2", model_client=model_client, system_message="You are a helpful assistant.")
+    inner_termination = MaxMessageTermination(2)  # Reduce to 2 to use fewer responses
+    inner_team = RoundRobinGroupChat([agent1, agent2], termination_condition=inner_termination, runtime=runtime)
+
+    # Test 1: Test team with output_task_messages=True (default behavior)
+    messages_with_task: List[BaseAgentEvent | BaseChatMessage] = []
+    async for message in inner_team.run_stream(task="Count to 10", output_task_messages=True):
+        if not isinstance(message, TaskResult):
+            messages_with_task.append(message)
+
+    # Should include the task message
+    assert len(messages_with_task) >= 1
+    assert any(
+        isinstance(msg, TextMessage) and msg.source == "user" and "Count to 10" in msg.content
+        for msg in messages_with_task
+    )
+
+    # Reset team before next test
+    await inner_team.reset()
+
+    # Test 2: Test team with output_task_messages=False
+    messages_without_task: List[BaseAgentEvent | BaseChatMessage] = []
+    async for message in inner_team.run_stream(task="Count to 10", output_task_messages=False):
+        if not isinstance(message, TaskResult):
+            messages_without_task.append(message)
+
+    # Should NOT include the task message in the stream
+    assert not any(
+        isinstance(msg, TextMessage) and msg.source == "user" and "Count to 10" in msg.content
+        for msg in messages_without_task
+    )
+
+    # Reset team before next test
+    await inner_team.reset()
+
+    # Test 3: Test SocietyOfMindAgent uses output_task_messages=False internally
+    # Create a separate model client for SocietyOfMindAgent to ensure we have enough responses
+    soma_model_client = ReplayChatCompletionClient(
+        ["Final response from society of mind"],
+    )
+    society_of_mind_agent = SocietyOfMindAgent("society_of_mind", team=inner_team, model_client=soma_model_client)
+
+    # Collect all messages from the SocietyOfMindAgent stream
+    soma_messages: List[BaseAgentEvent | BaseChatMessage] = []
+    async for message in society_of_mind_agent.run_stream(task="Count to 10"):
+        if not isinstance(message, TaskResult):
+            soma_messages.append(message)
+
+    # The SocietyOfMindAgent should output the task message (since it's the outer agent)
+    # but should NOT forward the task messages from its inner team
+    task_messages_in_soma = [msg for msg in soma_messages if isinstance(msg, TextMessage) and msg.source == "user"]
+
+    # Count how many times "Count to 10" appears in the stream
+    # With proper implementation, it should appear exactly once (from outer level only)
+    count_task_messages = sum(
+        1
+        for msg in soma_messages
+        if isinstance(msg, TextMessage) and msg.source == "user" and "Count to 10" in msg.content
+    )
+
+    # Should have exactly one task message (from the outer level only)
+    assert len(task_messages_in_soma) == 1
+    assert count_task_messages == 1  # Should appear exactly once, not duplicated from inner team
+
+    # Should have the SocietyOfMindAgent's final response
+    soma_responses = [msg for msg in soma_messages if isinstance(msg, TextMessage) and msg.source == "society_of_mind"]
+    assert len(soma_responses) == 1
+
+
 @pytest.mark.asyncio
 async def test_society_of_mind_agent_empty_messges(runtime: AgentRuntime | None) -> None:
     model_client = ReplayChatCompletionClient(
@@ -195,3 +273,49 @@ async def _mock_create(
     inner_team = RoundRobinGroupChat([agent1, agent2], termination_condition=inner_termination, runtime=runtime)
     society_of_mind_agent = SocietyOfMindAgent("society_of_mind", team=inner_team, model_client=model_client_soma)
     await society_of_mind_agent.run(task="Count to 10.")
+
+
+@pytest.mark.asyncio
+async def test_default_output_task_messages_behavior() -> None:
+    """Test that task messages are included by default (backward compatibility)."""
+    # Create inner team
+    model_client = ReplayChatCompletionClient(["Hello", "World", "TERMINATE"])
+    agent1 = AssistantAgent("agent1", model_client=model_client)
+    agent2 = AssistantAgent("agent2", model_client=model_client)
+    termination = TextMentionTermination("TERMINATE")
+    inner_team = RoundRobinGroupChat(participants=[agent1, agent2], termination_condition=termination)
+
+    streamed_messages: List[BaseAgentEvent | BaseChatMessage] = []
+    final_result: TaskResult | None = None
+
+    # Test default behavior (should include task messages since default is True)
+    async for message in inner_team.run_stream(task="Test default behavior"):
+        if isinstance(message, TaskResult):
+            final_result = message
+        else:
+            streamed_messages.append(message)
+
+    # Verify default behavior: task message should be included in stream
+    assert final_result is not None
+    task_message_found_in_stream = any(
+        isinstance(msg, TextMessage) and msg.source == "user" and "Test default behavior" in msg.content
+        for msg in streamed_messages
+    )
+    assert task_message_found_in_stream, "Task message should be included in stream by default"
+
+    # Validate that task message is included in the TaskResult.messages by default
+    task_message_in_result = any(
+        isinstance(msg, TextMessage) and msg.source == "user" and "Test default behavior" in msg.content
+        for msg in final_result.messages
+    )
+    assert task_message_in_result, "Task message should be included in TaskResult.messages by default"
+
+    # Verify the content structure makes sense (task message + agent responses)
+    user_messages = [msg for msg in final_result.messages if isinstance(msg, TextMessage) and msg.source == "user"]
+    agent_messages = [
+        msg for msg in final_result.messages if isinstance(msg, TextMessage) and msg.source in ["agent1", "agent2"]
+    ]
+
+    assert len(user_messages) >= 1, "Should have at least one user message (the task)"
+    assert len(agent_messages) >= 1, "Should have at least one agent response"
+    assert user_messages[0].content == "Test default behavior", "First user message should be the task"


Code review comments:
diff_hunk: @@ -504,7 +511,18 @@ async def stop_runtime() -> None:
             # Collect the output messages in order.
             output_messages: List[BaseAgentEvent | BaseChatMessage] = []
             stop_reason: str | None = None
-            # Yield the messsages until the queue is empty.
+            task_messages_list: List[BaseChatMessage] = []
+
+            # Add task messages to output_messages and stream only if output_task_messages is True
+            # Also track the actual task message objects for later filtering
+            if messages is not None:
+                for msg in messages:
+                    task_messages_list.append(msg)  # Track actual message objects
+                    if output_task_messages:
+                        output_messages.append(msg)  # Only add to output_messages if output_task_messages is True
path: python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_base_group_chat.py
text: @user1:
We don't need this anymore

diff_hunk: @@ -105,8 +107,11 @@ async def handle_start(self, message: GroupChatStart, ctx: MessageContext) -> No
                 GroupChatStart(messages=message.messages),
                 topic_id=DefaultTopicId(type=self._output_topic_type),
             )
-            for msg in message.messages:
-                await self._output_message_queue.put(msg)
+
+            # Only put messages in output queue if they are not task messages or output_task_messages is True
+            if self._output_task_messages:
path: python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_base_group_chat_manager.py
text: @user1:
Instead of constructor, use the GroupChatStart event to pass in the argument

diff_hunk: @@ -416,6 +421,8 @@ async def main() -> None:
             asyncio.run(main())
 
         """
+        # Update the output_task_messages field with the provided value
+        self._output_task_messages = output_task_messages
path: python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_base_group_chat.py
text: @user1:
No need for this