Python: [Bug]: WorkflowAgent.as_agent() echoes user input when HandoffBuilder internal executors emit events

[petetakacs]

Description

What happened?
When using HandoffBuilder with .as_agent() to create a workflow agent, the user's input message is echoed as the first part of the assistant's streamed response. This causes the final response to appear as a concatenation of the user's message followed by the actual agent response.

For example, if the user sends "Hi! Can you help me with something?", the streamed response appears as:
Hi! Can you help me with something?Hello! I'd love to help you with something...

What did you expect to happen?
The streamed response should only contain the assistant's actual response, not the user's input prefixed to it.

Steps to reproduce the issue
Create a workflow using HandoffBuilder with a coordinator and specialist agents
Wrap the workflow with .as_agent() to create a WorkflowAgent
Call run_stream() on the agent and observe the AgentRunUpdateEvent events
The first AgentRunUpdateEvent contains the user's input message as content
This happens because the handoff-coordinator executor (of type _HandoffCoordinator) emits this event, and it is NOT an AgentExecutor instance

Root Cause
The filtering logic in _convert_workflow_event_to_agent_update() (file: python/src/agent_framework_core/_workflows/_agent.py, lines 292-306) only checks for AgentExecutor instances when determining whether to suppress output:

if isinstance(executor, AgentExecutor):
    if executor.output_response == False:
        # Suppress

However, HandoffBuilder creates internal executors (_HandoffCoordinator, _InputToConversation) that inherit from BaseExecutor but are NOT AgentExecutor instances. These executors bypass the output_response filter entirely, causing user input to be echoed into the response stream.

Code Sample

from agent_framework import HandoffBuilder, ChatAgent

# Create specialist agents
something_x_agent = ChatAgent(
    name="something_x_agent",
    instructions="You help help me with something x",
    chat_client=...,
)

something_y_agent = ChatAgent(
    name="something_y_agent", 
    instructions="You generate something y",
    chat_client=...,
)

# Build workflow with HandoffBuilder
workflow = (
    HandoffBuilder()
    .with_autonomous_mode(turn_limit=1)
    .with_coordinator(
        instructions="Route to specialists based on user request",
        chat_client=...,
        name="coordinator",
    )
    .add_participant(something_x_agent)
    .add_participant(something_y_agent)
    .build()
)

# Create agent from workflow
agent = workflow.as_agent()

# Run stream - user input gets echoed!
async for event in agent.run_stream("Hi, help me with something y"):
    if isinstance(event, AgentRunUpdateEvent):
        print(event.content)  # First event contains the user's input!

Error Messages / Stack Traces

Error Messages / Stack Traces:
No error is thrown - this is a logic bug causing incorrect behavior in the stream output.

Package Versions

agent-framework-core: 1.0.0b260107, agent-framework-ag-ui: 1.0.0b260107

Python Version

3.12

Additional Context

This bug significantly impacts AG-UI integrations where the DefaultOrchestrator accumulates all streamed text. The echoed user input becomes part of the MESSAGES_SNAPSHOT assistant message content, corrupting the conversation history displayed to users.

Suggested Fix
Extend the filtering logic in _convert_workflow_event_to_agent_update() to handle non-AgentExecutor internal executors created by HandoffBuilder:

# Option 1: Also check for internal executor types
if isinstance(executor, (AgentExecutor, _HandoffCoordinator, _InputToConversation)):
    if getattr(executor, 'output_response', None) == False:
        return None

# Option 2: Check for a flag on any executor type
if hasattr(executor, 'output_response') and executor.output_response == False:
    return None

Alternatively, the internal executors created by HandoffBuilder could be modified to not emit user input as AgentRunUpdateEvent content in the first place.

Workaround
I implemented a filter wrapper that suppresses the first TextMessageContentEvent if it matches the user's input, and also fixes MessagesSnapshotEvent by removing the echoed prefix from assistant message content.

[moonbox3]

This should be solved in #3023. @TaoChenOSU, could you confirm:

HandoffBuilder creates internal executors (_HandoffCoordinator, _InputToConversation) that inherit from BaseExecutor but are NOT AgentExecutor instances.

This is not the case now in the new code?

[petetakacs]

I have tried version python-1.0.0b260114 but it is still present.