[Bug] single_turn LlmAgent node inputs (incl. inline binary) leak into the root agent LLM request on sequential multi-tool turns (workflow under a chat agent, branch=None)

Start in workflow/_llm_agent_wrapper.py at prepare_llm_agent_input and then inspect _is_event_belongs_to_branch and _should_include_event_in_context in flows/llm_flows/contents.py. Run the supplied repro.py to observe the root model requests across sequential tool calls. Done means single_turn node-input text and inline binary parts remain visible to the node but never appear in the root agent's requests.

Describe the Bug

When a Workflow containing single_turn LlmAgent nodes is run from a root chat agent's tool (tool_context.run_node(workflow, ...)), the agents' node-input user events leak into the ROOT agent's LLM request — including inline binary parts (PDF bytes) — during sequential multi-tool turns (the root calls tool A, sees its result, then calls tool B in the same user turn).

The leaked events are never persisted to the session; the pollution is in the in-memory event list only. But it directly inflates and cross-contaminates the root's request.

Root cause (traced in 2.5.0 source, each step verified)

1. prepare_llm_agent_input (workflow/_llm_agent_wrapper.py:203) gives each single_turn agent node a "private" session via session.model_copy(deep=False) and appends the node-input user event to it. But pydantic's shallow copy shares the same events list object — an append on the copy is an append on the original:

   s2 = s1.model_copy(deep=False); s2.events is s1.events  # True
   

2. The appended events carry branch=None, isolation_scope=None (verified by dumping the live events). Per _is_event_belongs_to_branch and _should_include_event_in_context (flows/llm_flows/contents.py), a falsy branch + None scope makes an event visible to every agent — including the root chat agent.

3. They are never persisted (the session service stores only real events; the persisted session holds 0 node-input events) — the leak exists only in the shared in-memory list.

4. Why only the SECOND in-turn tool call sees it: _rearrange_events_for_latest_function_response (contents.py:190) strips events between the current function_call and its function_response. The first workflow's node-input events sit inside that fc→fr window, so the root's round-1 request drops them. On the root's next round (the second tool call), they sit before the new fc — outside the current window — so they survive, verbatim, inline parts included, into the root's request.

This is the same root cause as #5989 (single_turn node inputs appended branchless/scopeless to the shared session.events), but a different symptom and a different trigger — and it survives the #6008 fix ("scope single-turn node inputs to workflow branch"): in the common topology of a workflow run from a root chat agent's tool, the workflow branch is None, so scoping to the workflow branch scopes to nothing.

Steps to Reproduce

1. pip install "google-adk==2.5.0" pypdf
2. Save the script below as repro.py.
3. python repro.py → prints per-round leak counts and LEAK REPRODUCED.

No network / API keys — both models are canned BaseLlm fakes.

Expected Behavior

Node inputs delivered to a single_turn agent node should be visible to that node's LLM request ONLY. They should never appear in any other agent's request, in any call pattern (parallel, sequential-in-turn, or across turns).

Observed Behavior

The root agent's second in-turn LLM request contains the worker node's input text and its inline PDF part verbatim (see repro output below), on google-adk 2.5.0.

root round 0: contents=1 leaked_worker_texts=0 leaked_inline_parts=0
root round 1: contents=3 leaked_worker_texts=0 leaked_inline_parts=0
root round 2: contents=7 leaked_worker_texts=1 leaked_inline_parts=1
LEAK REPRODUCED

Minimal Reproduction Code

"""Minimal repro: single_turn LlmAgent node inputs leak into the ROOT
agent's LLM request during sequential multi-tool turns. No network, no
API keys — the models are canned fakes.

Run:  python repro.py
Expected (buggy): ROUND 2 request contains the worker's input contents
(text + inline PDF bytes).
"""
from __future__ import annotations

import asyncio
import io

import pydantic
from google.adk import Agent
from google.adk.apps import App
from google.adk.artifacts import InMemoryArtifactService
from google.adk.models import BaseLlm
from google.adk.models.llm_request import LlmRequest
from google.adk.models.llm_response import LlmResponse
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import ToolContext
from google.adk.workflow import Workflow, node
from google.genai import types
from pypdf import PdfReader, PdfWriter


def _make_pdf(pages: int) -> bytes:
    writer = PdfWriter()
    for _ in range(pages):
        writer.add_blank_page(width=200, height=200)
    buf = io.BytesIO()
    writer.write(buf)
    return buf.getvalue()


class FakeModel(BaseLlm):
    """Canned model: emits the scripted tool calls one per round, then text."""

    model_config = pydantic.ConfigDict(extra="allow", arbitrary_types_allowed=True)

    def __init__(self, plan):
        super().__init__(model="fake")
        self.plan = plan
        self.n = 0
        self.requests: list[LlmRequest] = []

    async def generate_content_async(self, llm_request, stream=False):
        self.requests.append(llm_request)
        reply = self.plan[min(self.n, len(self.plan) - 1)]
        self.n += 1
        if isinstance(reply, dict):  # tool call
            parts = [
                types.Part(
                    function_call=types.FunctionCall(
                        name=reply["name"], args=reply["args"]
                    )
                )
            ]
        else:
            parts = [types.Part(text=reply)]
        yield LlmResponse(content=types.Content(role="model", parts=parts))


# The workflow: one stateless single_turn agent node whose input carries
# an inline PDF part — exactly how a binary payload is fed to an agent node.
worker_agent = Agent(
    model=FakeModel(["batch summary text"]),
    name="worker",
    description="stub",
    include_contents="none",
    mode="single_turn",
    instruction="stub",
)


@node(name="run_worker", rerun_on_resume=True)
async def run_worker(ctx, node_input: str):
    summary = await ctx.run_node(
        worker_agent,
        node_input=types.Content(
            role="user",
            parts=[
                types.Part(text="Please analyze the content of pages 1 to 10."),
                types.Part.from_bytes(
                    data=_make_pdf(10), mime_type="application/pdf"
                ),
            ],
        ),
    )
    return summary


workflow = Workflow(name="wf", edges=[("START", run_worker)])


async def the_tool(tool_context: ToolContext, label: str) -> dict:
    """Tool that runs the workflow — the root agent calls it twice in one turn."""
    out = await tool_context.run_node(
        workflow, node_input=label, run_id=f"run-{label}"
    )
    return {"status": "success", "content": str(out)[:40]}


async def main() -> None:
    # Root: call the tool for doc A, see the result, then call it for doc B
    # — sequential tool calls inside ONE user turn (a normal model pattern).
    root_plan = [
        {"name": "the_tool", "args": {"label": "a"}},
        {"name": "the_tool", "args": {"label": "b"}},
        "done",
    ]
    root_model = FakeModel(root_plan)
    app = App(
        name="repro",
        root_agent=Agent(
            name="root",
            model=root_model,
            instruction="help",
            tools=[the_tool],
        ),
    )
    runner = Runner(
        app=app,
        session_service=InMemorySessionService(),
        artifact_service=InMemoryArtifactService(),
        auto_create_session=True,
    )
    async for _ in runner.run_async(
        user_id="u",
        session_id="s",
        new_message=types.Content(
            role="user", parts=[types.Part(text="run both")]
        ),
    ):
        pass

    for i, req in enumerate(root_model.requests):
        leaked_texts = [
            p.text
            for c in (req.contents or [])
            for p in (c.parts or [])
            if p.text and p.text.startswith("Please analyze the content of pages")
        ]
        leaked_inline = sum(
            1 for c in (req.contents or []) for p in (c.parts or []) if p.inline_data
        )
        print(
            f"root round {i}: contents={len(req.contents or [])} "
            f"leaked_worker_texts={len(leaked_texts)} "
            f"leaked_inline_parts={leaked_inline}"
        )
    verdict = any(
        p.inline_data
        for req in root_model.requests[1:]
        for c in (req.contents or [])
        for p in (c.parts or [])
    )
    print("LEAK REPRODUCED" if verdict else "clean")


asyncio.run(main())

Environment Details

• ADK Library Version: google-adk 2.5.0
• Desktop OS: macOS 26.4.1 (repro has no OS-specific code)
• Python Version: 3.14.0

Additional Context

Workarounds considered:

• An agent-level before_model_callback on the root that fingerprint-strips the node-input contents (by their known text prefixes / unexpected inline parts) — works but is a band-aid over what should be a structural guarantee.
• Related: #5989 / #6008 fixed the sibling-worker crash flavor of this same root cause; the parent-agent-request leak flavor remains.