A task-mode root agent loses its task input once it asks the user a question #7384

🔴 Required Information

Describe the Bug:

Runner.run_async documents a root LlmAgent(mode="task") as fully supported, and it is the server shape the RemoteA2aAgent task-mode guide puts behind to_a2a. Two things go wrong as soon as that root asks the user something before calling finish_task.

1. On the turn after the question, the model no longer sees the original request. It sees the user's answer twice: once as the task input at the head of the contents, once in the history.
2. A later task in the same session runs under the same isolation scope (<agent>@1), which _find_active_task_scope already counts as finished. The user's answer to a question in that later task is therefore not stamped with the scope, and it reaches the model only as the task input at the head of the contents. The history the model sees ends on its own question.

Steps to Reproduce:

1. pip install google-adk==2.11.0
2. Run the script below. It needs no A2A and no API key: a scripted model and an in-memory session.

import asyncio
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.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from pydantic import Field


class ScriptedLlm(BaseLlm):
    script: list[Any] = Field(default_factory=list)
    requests: list[Any] = Field(default_factory=list)

    async def generate_content_async(self, llm_request, stream=False):
        self.requests.append(llm_request)
        line = self.script.pop(0)
        if isinstance(line, str):
            part = types.Part(text=line)
        else:
            part = types.Part(function_call=types.FunctionCall(name=line[0], args=line[1]))
        yield LlmResponse(content=types.Content(role="model", parts=[part]))


def seen(request):
    return [
        (c.role, p.text or (p.function_call and f"call {p.function_call.name}")
         or (p.function_response and f"response {p.function_response.name}") or "")
        for c in request.contents
        for p in c.parts
    ]


async def main():
    llm = ScriptedLlm(model="scripted", script=[
        "Which period?",
        ("finish_task", {"result": "Q3 revenue by country"}),
        "Which devices?",
        ("finish_task", {"result": "Q3 revenue by device"}),
    ])
    agent = LlmAgent(name="reporting", model=llm, instruction="Run reports.", mode="task")
    sessions = InMemorySessionService()
    runner = Runner(app_name="app", agent=agent, session_service=sessions)
    await sessions.create_session(app_name="app", user_id="u", session_id="s")
    for text in ["Revenue by country", "Q3", "Same thing by device", "Desktop and mobile"]:
        async for _ in runner.run_async(
            user_id="u", session_id="s",
            new_message=types.Content(role="user", parts=[types.Part(text=text)]),
        ):
            pass
    for index, request in enumerate(llm.requests):
        print(f"model call {index}: {seen(request)}")


asyncio.run(main())

Expected Behavior:

• Call 1: [('user', 'Revenue by country'), ('model', 'Which period?'), ('user', 'Q3')].
• Call 3: "Same thing by device" as the task input, and a history that ends on ('user', 'Desktop and mobile'), after ('model', 'Which devices?').

Observed Behavior:

model call 0: [('user', 'Revenue by country')]
model call 1: [('user', 'Q3'), ('model', 'Which period?'), ('user', 'Q3')]
model call 2: [('user', 'Same thing by device'), ('model', 'Which period?'), ('user', 'Q3'), ('model', 'call finish_task'), ('user', 'response finish_task')]
model call 3: [('user', 'Desktop and mobile'), ('model', 'Which period?'), ('user', 'Q3'), ('model', 'call finish_task'), ('user', 'response finish_task'), ('model', 'Which devices?')]

Environment Details:

• ADK Library Version (pip show google-adk): 2.11.0. src/google/adk/flows/llm_flows/context/_contents.py is unchanged on main.
• Desktop OS: macOS
• Python Version (python -V): 3.12.12

Model Information:

• Are you using LiteLLM: No, the reproduction uses a scripted BaseLlm. Production goes through LiteLLM.
• Which model is being used: N/A for the reproduction (Claude Sonnet in production)

🟡 Optional Information

Regression:

Not known. A root LlmAgent(mode="task") is what the reproduction exercises; it was not tried on earlier versions.

Where it comes from:

• The first user message of a task root is appended before the task opens, so it carries no isolation_scope, and the contents filter for scope reporting@1 drops it.
• _build_task_input_user_content then looks for a function call whose id is the scope, finds none (a root has no delegating call), and falls back to invocation_context.user_content. When a message continues a paused task, _node_runner_utils deliberately leaves user_content as the new message ("A message that joins a paused task only borrowed that invocation's id"), so the fallback is the answer, not the request.
• Every invocation of the root runs as node reporting@1, so every task in the session shares one scope, and once one task has called finish_task that scope stays in finished_scopes.

Impact:

Any multi-turn task served as a root, which includes every to_a2a server a RemoteA2aAgent(mode="task") delegates to: the remote agent asks a clarifying question, and on the next turn it no longer knows what it was asked to do.

A chat coordinator that delegates to the same agent as a task sub-agent does not have the problem, because the delegating function call is in the session and its id is the scope.