RemoteA2aAgent: a terminal A2A task carrying tool activity fails is_final_response(), so the turn cannot be detected as finished (non-streaming peer)

[ferponse]

Environment

• google-adk 2.6.0
• a2a-sdk 1.1.2
• Python 3.12

Summary

When a RemoteA2aAgent talks to a peer that does not advertise streaming, the a2a-sdk
correctly falls back to message/send and ADK hands the whole response over as a single
Event carrying the narration text, every function_call, every function_response and the
closing text.

That event's task state is terminal — event.custom_metadata["a2a:response"]["status"]["state"]
is TASK_STATE_COMPLETED — but event.is_final_response() returns False, because the
predicate excludes any event holding tool activity:

# google/adk/events/event.py
def is_final_response(self) -> bool:
    if self.actions.skip_summarization or self.long_running_tool_ids:
        return True
    return (
        not self.get_function_calls()
        and not self.get_function_responses()
        and not self.partial
        and not self.has_trailing_code_execution_result()
    )

So ADK produces a completed task in one event, and its own public helper cannot recognise that
event as the end of the turn. A consumer that closes the turn on is_final_response() — which is
what the docstring recommends ("Application and UI layers can rely on this helper to detect a
complete, user-facing response instead of replicating its logic") — never closes it.

This is the default path for a peer built with ADK, not an exotic one: to_a2a(...) builds
its card through AgentCardBuilder with capabilities.streaming left at False, so an
ADK-served agent is non-streaming unless the card is built by hand.

Reproduction

server.py — a plain ADK agent served over A2A, with the default card:

from google.adk.a2a.executor.a2a_agent_executor import A2aAgentExecutor
from google.adk.a2a.utils.agent_to_a2a import to_a2a
from google.adk.agents import LlmAgent

def divide(a: float, b: float) -> dict:
    return {"error": "cannot divide by zero"} if b == 0 else {"quotient": a / b}

root_agent = LlmAgent(name="peer", model="<any>", tools=[divide],
                      instruction="Use the tool, then summarise the result.")

# force_new_version so the v2 integration path is exercised; the v1 path behaves the same way.
app = to_a2a(root_agent, host="127.0.0.1", port=8100,
             agent_executor_factory=lambda runner: A2aAgentExecutor(
                 runner=runner, use_legacy=False, force_new_version=True))

Client side: a RemoteA2aAgent pointing at http://127.0.0.1:8100, as a sub-agent of any root
agent, and one turn that makes the peer call its tool (e.g. "ask the peer to divide 10 by 0").

Observed

The peer's card says {"streaming": false}, so a2a/client/base_client.py takes the
non-streaming branch:

if not self._config.streaming or not self._card.capabilities.streaming:
    response = await self._execute_with_interceptors(..., method='send_message', ...)

message/send returns one completed task with three artifacts (text, function_call,
function_response), and _handle_a2a_response_v2 converts it into one Event where:

• event.get_function_calls() and event.get_function_responses() are both non-empty,
• event.custom_metadata["a2a:response"]["status"]["state"] == "TASK_STATE_COMPLETED",
• event.is_final_response() is False,
• and it is the last event of the invocation — run_async ends right after it.

Expected

Either of these would let a consumer detect the end of the turn using ADK's own API:

1. is_final_response() recognises an event whose A2A task state is terminal (regardless of the
   tool activity it carries) — for instance by having the handler set actions.skip_summarization
   or an equivalent marker when the task state is terminal; or
2. _handle_a2a_response_v2 emits one event per artifact for a non-streaming response, mirroring
   what the streaming path produces, so the last event satisfies the predicate.

Option 2 has a second benefit: today the same logical turn is persisted with a different shape
depending only on whether the peer streams (one event with N parts vs. N events), which any
consumer storing events has to absorb.

Impact / workaround

A UI driven by is_final_response() shows the tool activity and then waits forever, since nothing
marks the turn as over. The workaround is to stop trusting the helper for A2A sub-agents and read
the task state directly:

closes_task = (
    event.author in remote_a2a_agent_names
    and task_state_of(event) in TERMINAL_STATES  # from custom_metadata["a2a:response"]
)
is_final = event.is_final_response() or closes_task

which means every consumer has to re-implement the part of the predicate ADK asks them not to.

Related: #6274 (a2a-sdk 1.x), #4016 (is_final_response() docs vs. code).

[ferponse]

For maintainers triaging both: this and #6585 are the two ways an A2A turn can end without a detectable close — one with a terminal task the API does not recognise as final, the other with no terminal task at all. They are independent and either can be fixed alone.

[Linxiushen]

I'd like to work on a focused fix for this. My first step would be to trace the v2 response conversion and identify the narrowest existing event signal for a terminal A2A task, then add unit coverage for a completed non-streaming response that contains both function calls/responses and final text. I would avoid changing the generic Event.is_final_response() contract unless the A2A layer has no appropriate way to mark the terminal event.

Please let me know if that direction is welcome, and I can prepare the patch.

[surajksharma07]

Reproduced this on 2.6.2 — convert_a2a_task_to_event never reflects the terminal task state into event.actions.skip_summarization, so is_final_response() falls through on any completed task carrying tool activity.

Your proposed direction checks out — setting skip_summarization when the state is completed/failed/canceled fixes it locally without touching the generic contract.
Could you try that on your branch and confirm it holds (including the input-required path) before putting up the PR?