chat wrapper (run_llm_agent_as_node) extracts task-delegation FCs from partial=True progressive-SSE chunks and breaks — the FC event is never persisted, orphaning the synthesized task FR and permanently poisoning the session

[kenwilly]

Environment

• Python 3.13
• google-adk==2.0.0
• Default ADK feature: PROGRESSIVE_SSE_STREAMING (on by default)
• Triggered under RunConfig(streaming_mode=StreamingMode.SSE)

Bug description

Under RunConfig(streaming_mode=SSE), ADK's StreamingResponseAggregator runs in
PROGRESSIVE_SSE_STREAMING mode (default-on) and marks every intermediate chunk
partial=True — including the chunk that carries a task-delegation FunctionCall. The
non-partial aggregate that re-carries the FC is only produced by aggregator.close()
after the model stream fully drains.

ADK's chat-mode wrapper, google.adk.workflow._llm_agent_wrapper.run_llm_agent_as_node,
extracts task-delegation FCs from every yielded event with no partial check,
dispatches the specialist, and breaks out of the agent's event generator. Two defects
follow:

1. The FC event is never persisted. The Runner only appends events with
   partial is not True to the session; the break closes the generator before the
   non-partial aggregate is ever yielded. The synthesized task FunctionResponse is
   persisted — so the very next contents build sees an orphaned FR (a function response
   with no matching function call) and raises
   ValueError: No function call event found for function responses ids: {...}. The
   session is permanently poisoned: every later turn replays the same history and
   re-raises.
2. Dispatch can fire with truncated args. Progressive SSE streams FC arguments
   across chunks; extracting from the first partial chunk that names the FC can dispatch
   the specialist with incomplete (even empty) input.

This defect is the dominant producer of the orphaned-function-response poisoned
state reported in the companion draft
docs/upstream-reports/02-orphaned-function-response-poisoning.md: the wrapper
synthesizes and persists a task-delegation function response while never persisting
the matching function call.

Reproduction

A self-contained Python repro drives a real Runner + coordinator(mode='chat') +
task specialist whose stub coordinator LLM streams the delegation FC in the exact
progressive-SSE shape (a partial=True chunk first, then the partial=False aggregate
re-carrying the same FC with the same id). No downstream workaround is installed — this
is the raw ADK 2.0.0 behaviour.

"""
Minimal repro: ADK 2.0.0 progressive-SSE task-dispatch defect.

Run:  cd <repo with google-adk==2.0.0> && python repro_partial_sse.py
Expected: the coordinator's re-entry contents build raises the
orphaned-response ValueError; the persisted session shows the task FR with no
matching task FC.
"""

from __future__ import annotations

import asyncio
import traceback
from collections.abc import AsyncIterator
from typing import Any

from google.adk.agents import LlmAgent
from google.adk.models.base_llm import BaseLlm
from google.adk.models.llm_response import LlmResponse
from google.genai.types import Content, FunctionCall, Part

_COORDINATOR_CALLS: list[int] = []
_SPECIALIST_CALLS: list[int] = []


class _StreamingDispatchCoordinatorLlm(BaseLlm):
    """Coordinator stub emitting the progressive-SSE dispatch shape.

    Call 1 yields the task FC twice, exactly as ADK's StreamingResponseAggregator
    does under PROGRESSIVE_SSE_STREAMING: a partial=True chunk first, then the
    partial=False aggregate re-carrying the same FC (same id).
    """

    model: str = "streaming_dispatch_stub"

    @classmethod
    def supported_models(cls) -> list[str]:
        return ["streaming_dispatch_stub"]

    async def generate_content_async(  # type: ignore[override]
        self, llm_request: Any, stream: bool = False
    ) -> AsyncIterator[LlmResponse]:
        _COORDINATOR_CALLS.append(1)
        if len(_COORDINATOR_CALLS) == 1:
            fc = FunctionCall(
                name="ga_specialist",
                args={"request": "Top landing pages for June 2026"},
                id="fc-dispatch-1",
            )
            # 1. partial chunk carrying the task FC — every intermediate
            #    progressive-SSE chunk is marked partial=True.
            yield LlmResponse(
                content=Content(role="model", parts=[Part(function_call=fc)]),
                partial=True,
            )
            # 2. non-partial aggregate re-carrying the same FC (same id) —
            #    only this is persisted by the Runner.
            yield LlmResponse(
                content=Content(role="model", parts=[Part(function_call=fc)]),
                partial=False,
                turn_complete=True,
            )
        else:
            yield LlmResponse(
                content=Content(
                    role="model",
                    parts=[Part.from_text(text="Here are your top landing pages.")],
                ),
                partial=False,
                turn_complete=True,
            )


class _FinishingSpecialistLlm(BaseLlm):
    """Task specialist stub: immediately calls finish_task with its result."""

    model: str = "finishing_specialist_stub"

    @classmethod
    def supported_models(cls) -> list[str]:
        return ["finishing_specialist_stub"]

    async def generate_content_async(  # type: ignore[override]
        self, llm_request: Any, stream: bool = False
    ) -> AsyncIterator[LlmResponse]:
        _SPECIALIST_CALLS.append(1)
        fc = FunctionCall(
            name="finish_task",
            args={"result": "1. /pricing 2. /blog 3. /home"},
            id=f"fin-{len(_SPECIALIST_CALLS)}",
        )
        yield LlmResponse(
            content=Content(role="model", parts=[Part(function_call=fc)]),
            turn_complete=True,
        )


async def _run_repro() -> None:
    from google.adk.runners import Runner
    from google.adk.sessions import InMemorySessionService
    from google.genai import types as genai_types

    specialist = LlmAgent(
        name="ga_specialist",
        model=_FinishingSpecialistLlm(),
        mode="task",
        instruction="GA specialist",
        disallow_transfer_to_parent=True,
    )
    coordinator = LlmAgent(
        name="coordinator",
        model=_StreamingDispatchCoordinatorLlm(),
        mode="chat",
        instruction="Coordinator",
        tools=[],
        sub_agents=[specialist],
    )

    session_service = InMemorySessionService()
    session = await session_service.create_session(
        app_name="partial_dispatch_repro", user_id="test_user"
    )
    runner = Runner(
        agent=coordinator,
        app_name="partial_dispatch_repro",
        session_service=session_service,
    )

    print("Running a progressive-SSE-shaped dispatch turn (raw ADK, no guard)...\n")
    try:
        async for event in runner.run_async(
            user_id=session.user_id,
            session_id=session.id,
            new_message=genai_types.Content(
                role="user",
                parts=[genai_types.Part.from_text(text="Top landing pages June 2026?")],
            ),
        ):
            pass
        print("NO RAISE — the defect may be fixed upstream; re-check on this ADK version.")
    except ValueError as exc:
        print("RAISED ValueError (orphaned task FR — the progressive-SSE poison):\n")
        print(f"  type: {type(exc).__module__}.{type(exc).__name__}")
        print(f"  message: {exc}\n")
        print("Traceback (key frames):")
        print("----")
        print(traceback.format_exc())
        print("----")

    persisted_session = await session_service.get_session(
        app_name="partial_dispatch_repro",
        user_id=session.user_id,
        session_id=session.id,
    )
    persisted = list(persisted_session.events) if persisted_session else []

    def _fc_ids(name: str) -> list[str]:
        out = []
        for e in persisted:
            for fc in e.get_function_calls():
                if fc.name == name:
                    out.append(fc.id)
        return out

    def _fr_ids(name: str) -> list[str]:
        out = []
        for e in persisted:
            for fr in e.get_function_responses():
                if fr.name == name:
                    out.append(fr.id)
        return out

    print("\n── Persisted session events (post-turn) ──")
    for i, e in enumerate(persisted):
        role = getattr(e.content, "role", "?") if e.content else "?"
        parts = []
        for p in (e.content.parts if e.content else []):
            if p.function_call:
                parts.append(f"FC:{p.function_call.name}(id={p.function_call.id})")
            elif p.function_response:
                parts.append(f"FR:{p.function_response.name}(id={p.function_response.id})")
            elif p.text:
                parts.append(f"text:{p.text[:40]!r}")
        print(f"  [{i}] role={role} partial={e.partial}  parts={parts}")

    fc_ids = _fc_ids("ga_specialist")
    fr_ids = _fr_ids("ga_specialist")
    print()
    print(f"Persisted coordinator task FC ids: {fc_ids}")
    print(f"Persisted synthesized task FR ids: {fr_ids}")
    print(f"Specialist dispatched {len(_SPECIALIST_CALLS)}x for one task FC")
    print()
    if not fc_ids and fr_ids:
        print(
            "BUG CONFIRMED: the coordinator's task-delegation FunctionCall event\n"
            "was NOT persisted (the wrapper broke out on the partial=True chunk,\n"
            "before the non-partial aggregate was ever yielded), while the\n"
            "synthesized task FunctionResponse WAS persisted — an orphaned FR.\n"
            "Every later contents build raises the orphaned-response ValueError.\n"
            "Defect 3 is the dominant producer of defect 2's poisoned state."
        )
    elif fc_ids and fr_ids:
        print(
            "No orphan — the FC was persisted. If this is ADK 2.0.0 with the\n"
            "guard removed, the defect may be fixed upstream; re-check."
        )


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

Observed output (run against google-adk==2.0.0, Python 3.13):

Running a progressive-SSE-shaped dispatch turn (raw ADK, no guard)...

RAISED ValueError (orphaned task FR — the progressive-SSE poison):

  type: builtins.ValueError
  message: No function call event found for function responses ids: {'fc-dispatch-1'}

Traceback (key frames):
----
  File ".../google/adk/workflow/_llm_agent_wrapper.py", line 380, in run_llm_agent_as_node
    async for event in run_iter:
      ...
        break
  File ".../google/adk/flows/llm_flows/base_llm_flow.py", line 935, in _preprocess_async
    async for event in agen:
      yield event
  File ".../google/adk/flows/llm_flows/contents.py", line 74, in run_async
    llm_request.contents = _get_contents(
  File ".../google/adk/flows/llm_flows/contents.py", line 637, in _get_contents
    result_events = _rearrange_events_for_latest_function_response(
        filtered_events
    )
  File ".../google/adk/flows/llm_flows/contents.py", line 224, in _rearrange_events_for_latest_function_response
    raise ValueError(
        ...
    )
ValueError: No function call event found for function responses ids: {'fc-dispatch-1'}

----

── Persisted session events (post-turn) ──
  [0] role=user partial=None  parts=["text:'Top landing pages June 2026?'"]
  [1] role=model partial=None  parts=['FC:finish_task(id=fin-1)']
  [2] role=user partial=None  parts=['FR:finish_task(id=fin-1)']
  [3] role=user partial=None  parts=['FR:ga_specialist(id=fc-dispatch-1)']
  [4] role=? partial=None  parts=[]

Persisted coordinator task FC ids: []
Persisted synthesized task FR ids: ['fc-dispatch-1']
Specialist dispatched 1x for one task FC

BUG CONFIRMED: the coordinator's task-delegation FunctionCall event
was NOT persisted (the wrapper broke out on the partial=True chunk,
before the non-partial aggregate was ever yielded), while the
synthesized task FunctionResponse WAS persisted — an orphaned FR.
Every later contents build raises the orphaned-response ValueError.
Defect 3 is the dominant producer of defect 2's poisoned state.

The persisted-event dump is the smoking gun: event [3] is FR:ga_specialist(id=fc-dispatch-1)
with no matching FC:ga_specialist anywhere in the session (Persisted coordinator task FC ids: []). The specialist dispatched exactly once, so the orphan is not a double
dispatch — it is the wrapper breaking on the partial=True chunk before the
partial=False aggregate (the only event the Runner persists) was ever yielded.

Root cause

google/adk/workflow/_llm_agent_wrapper.py, function _extract_task_delegation_fcs
(line 56) has no partial check:

def _extract_task_delegation_fcs(
    event: Event, tools_dict: dict
) -> list[types.FunctionCall]:
  """Return task-delegation FCs from this event."""
  from ..tools.agent_tool import _TaskAgentTool

  return [
      fc
      for fc in event.get_function_calls()        # ← no `event.partial` gate
      if fc.id
      and fc.name in tools_dict
      and isinstance(tools_dict[fc.name], _TaskAgentTool)
  ]

It is called from run_llm_agent_as_node (line 382) inside the loop that breaks on any
extracted task FC (line 388):

  async with aclosing(run_method) as run_iter:
    async for event in run_iter:
      yield event
      task_fcs = _extract_task_delegation_fcs(event, tools_dict)   # line 382
      for fc in task_fcs:
        output = await _dispatch_task_fc(agent, fc, ctx)
        yield _synthesize_task_fr_event(fc, output)
      if task_fcs:
        had_task_fc = True
        break   # line 388 — closes run_iter before the non-partial aggregate

Under progressive SSE, the first chunk carrying the task FC is partial=True. The wrapper
extracts from it, dispatches, synthesizes the task FR, and breaks. The Runner only
persists events with partial is not True, so:

• the synthesized task FunctionResponse (yielded by the wrapper itself, not by the
  inner generator's partial=True chunk) is persisted, but
• the coordinator's FunctionCall event is only ever carried by the partial=False
  aggregate, which the break never reached → it is never persisted.

The result is an orphaned function response, which the next _preprocess_async contents
build raises on (contents.py:224 via _get_contents at contents.py:637).

Expected behaviour

Task-delegation FC extraction should be gated on non-partial events, exactly as ADK's
own process_llm_agent_output does for the non-streamed path — dispatch from the final
aggregate, which the Runner persists and which carries the complete FC args. Concretely,
_extract_task_delegation_fcs (or its caller) should return no FCs for events whose
partial is True, so the break only fires on the non-partial aggregate.

Observed behaviour

Without the gate, the wrapper dispatches from the partial=True chunk, breaks before
the non-partial aggregate is yielded, and the session is left with a synthesized task FR
and no matching task FC — permanently poisoned by the orphaned-response raise documented
in the companion report.

Related

• Consumption defect (companion report): docs/upstream-reports/02-orphaned-function-response-poisoning.md
  — the orphaned-FR ValueError this defect produces is the same raise that turns any
  orphaned FR into a dead session. This report is the producer; that report is the
  consumer. Fixing the wrapper's partial gate removes the dominant producer; fixing the
  contents processor's fatal raise (companion) removes the fatal consumption.
• Mixed-turn FC drop (companion report): docs/upstream-reports/01-mixed-turn-fc-drop.md
  — the same run_llm_agent_as_node break (line 388) is also the root of the mixed-turn
  regular-FC drop. Both defects share the wrapper's break-on-task-FC design; a single
  "drain remaining events before break / gate on non-partial" fix sketch addresses
  both.