[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) #7227

"""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())