StreamingResponseAggregator causes infinite tool-call loop with transfer_to_agent + streaming

[brucearctor]

New Issue: StreamingResponseAggregator infinite loop in multi-agent workflows

Title: StreamingResponseAggregator causes infinite tool-call loop with transfer_to_agent + streaming

Labels: bug, streaming

---

Description

When using StreamingMode.SSE (the default) with multi-agent workflows that use transfer_to_agent, the StreamingResponseAggregator causes an infinite loop where the LLM repeatedly re-calls tools with slightly different arguments. The agent never exits its turn.

Setting StreamingMode.NONE fixes the issue — 3 LLM calls (correct) vs 20+ (infinite loop).

Minimal Reproduction

"""
Minimal repro: streaming + transfer_to_agent → infinite tool-call loop.

Requirements:
    pip install google-adk google-genai

Usage:
    GOOGLE_API_KEY=<key> python repro_streaming_loop.py
"""
import asyncio
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai.types import Content, Part
from google.adk.agents.run_config import RunConfig, StreamingMode


# ── Two trivial tools ──────────────────────────────────────────────
MENU = {
    "Margherita Pizza": 12.00,
    "Caesar Salad": 8.50,
    "Sparkling Water": 3.00,
}

def lookup_item(item_name: str) -> dict:
    """Look up a menu item by name. Returns price and availability."""
    price = MENU.get(item_name)
    if price:
        return {"item": item_name, "price": price, "available": True}
    return {"item": item_name, "available": False}

def add_to_cart(item_name: str, quantity: int) -> dict:
    """Add an item to the customer's cart."""
    price = MENU.get(item_name, 0)
    return {"added": item_name, "quantity": quantity, "subtotal": price * quantity}


# ── Two agents: orchestrator + sub-agent ───────────────────────────
menu_agent = Agent(
    name="menu_agent",
    model="gemini-2.0-flash",
    instruction="You help customers browse the menu. Use lookup_item to check items.",
    tools=[lookup_item],
)

orchestrator = Agent(
    name="orchestrator",
    model="gemini-2.0-flash",
    instruction=(
        "You are a food ordering assistant. "
        "Transfer to menu_agent for menu questions. "
        "Use add_to_cart to add items."
    ),
    tools=[add_to_cart],
    sub_agents=[menu_agent],
)


async def run_test(streaming_mode: StreamingMode) -> int:
    """Run one turn and count how many events are produced."""
    session_service = InMemorySessionService()
    runner = Runner(
        agent=orchestrator,
        app_name="repro",
        session_service=session_service,
    )
    session = await session_service.create_session(
        app_name="repro", user_id="test"
    )
    config = RunConfig(streaming_mode=streaming_mode)

    user_msg = Content(
        role="user",
        parts=[Part(text="What's on the menu? Can you look up Margherita Pizza?")],
    )

    event_count = 0
    async for event in runner.run_async(
        user_id="test",
        session_id=session.id,
        new_message=user_msg,
        run_config=config,
    ):
        event_count += 1
        if event_count > 50:
            print(f"  ⚠️  Stopping at {event_count} events (likely infinite loop)")
            break
        if event.content and event.content.parts:
            for part in event.content.parts:
                if hasattr(part, "function_call") and part.function_call:
                    print(f"  [{event_count}] function_call: {part.function_call.name}")
                elif part.text:
                    preview = part.text[:60].replace("\n", " ")
                    print(f"  [{event_count}] text: {preview}...")

    return event_count


async def main():
    print("=" * 60)
    print("Test 1: StreamingMode.SSE (default)")
    print("=" * 60)
    count_sse = await run_test(StreamingMode.SSE)
    print(f"→ Total events: {count_sse}\n")

    print("=" * 60)
    print("Test 2: StreamingMode.NONE")
    print("=" * 60)
    count_none = await run_test(StreamingMode.NONE)
    print(f"→ Total events: {count_none}\n")

    print("=" * 60)
    print("Summary")
    print("=" * 60)
    print(f"  SSE:  {count_sse} events {'⚠️ LIKELY LOOP' if count_sse > 30 else '✅'}")
    print(f"  NONE: {count_none} events {'⚠️ LIKELY LOOP' if count_none > 30 else '✅'}")


if __name__ == "__main__":
    asyncio.run(main())

Systematic Isolation

We also tested in a production Temporal context with 6 configurations:

#	Configuration	LLM Calls	Behavior	Status
1	StreamingMode.SSE (default)	57	Infinite aggregator loop — LLM keeps re-calling tools	❌
2	StreamingMode.SSE + PROGRESSIVE_SSE_STREAMING=1	57	Same infinite loop	❌
3	StreamingMode.NONE	18	Correct: 1 turn, proper handoff, clean exit	✅
4	StreamingMode.SSE + model override (non-streaming)	18	Works when model ignores stream flag	✅
5	StreamingMode.NONE + simulated UI streaming	18	Production workaround	✅
6	Direct Gemini API (no ADK)	N/A	Streaming works perfectly — ADK-layer issue	✅

Root Cause Analysis

The issue appears to be in StreamingResponseAggregator — specifically how it reassembles streamed function-call chunks:

1. The aggregator collects partial function_call parts from the stream
2. It merges them and re-emits as a complete LlmResponse
3. The merged response structure causes base_llm_flow to treat it as a new turn
4. The LLM sees the function call "echo" and generates another function call with slightly different args
5. This loops until max_llm_calls is hit or context overflows

With StreamingMode.NONE, the LLM response arrives as a single complete object — no aggregation needed — and the agent loop exits cleanly.

Workaround

Use StreamingMode.NONE and simulate streaming at the application layer:

config = RunConfig(streaming_mode=StreamingMode.NONE)

for event in runner.run(..., run_config=config):
    if event.content and event.content.parts:
        for part in event.content.parts:
            if part.text:
                full_text += part.text

# Simulate streaming to the UI
for i in range(0, len(full_text), chunk_size):
    publish_chunk(full_text[i:i+chunk_size])
    await asyncio.sleep(0.03)

Environment

• ADK: 2.6.0
• google-genai: 1.20.0
• Model: gemini-2.0-flash (also reproduced on gemini-2.5-flash, gemini-3.5-flash)
• Python: 3.13
• OS: macOS 15

Related to #3974 (closed).

[brucearctor]

I'm also working on a fix :-)

[brucearctor]

@wuliang229 -- it'd be great if you wanted to review this, though :-)