fix(a2a): send complete legacy text once - #7542

"""Offline Runner -> RemoteA2aAgent -> to_a2a regression probe."""
import argparse
import asyncio
import importlib.metadata
import json
from pathlib import Path
import sys

parser = argparse.ArgumentParser()
parser.add_argument('--source', type=Path)
parser.add_argument('--output', type=Path, required=True)
args = parser.parse_args()
if args.source:
    sys.path.insert(0, str(args.source.resolve()))

import httpx
import google.adk.a2a.executor.a2a_agent_executor as executor_module
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
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent
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

ANSWER = 'The answer is 42.'

class ScriptedModel(BaseLlm):
    async def generate_content_async(self, llm_request, stream=False):
        last = llm_request.contents[-1].parts[-1]
        if self.model == 'plain' or last.function_response:
            parts = [types.Part(text=ANSWER)]
        else:
            parts = [types.Part(text='Checking the value.'),
                     types.Part.from_function_call(name='lookup', args={})]
        yield LlmResponse(content=types.Content(role='model', parts=parts))

def model_texts(events):
    return [part.text for event in events
            if event.author == 'remote' and event.content
            and event.content.role == 'model'
            for part in event.content.parts or [] if part.text]

async def probe(use_v2, use_tool):
    calls = []
    def lookup() -> str:
        calls.append('lookup')
        return '42'
    def factory(runner):
        return A2aAgentExecutor(runner=runner, force_new_version=True)
    app = to_a2a(LlmAgent(name='remote',
        model=ScriptedModel(model='tool' if use_tool else 'plain'),
        tools=[lookup] if use_tool else []),
        agent_executor_factory=factory if use_v2 else None)
    async with app.router.lifespan_context(app):
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app)) as client:
            remote = RemoteA2aAgent(name='remote',
                agent_card='http://localhost:8000/.well-known/agent-card.json',
                httpx_client=client)
            sessions = InMemorySessionService()
            async with Runner(app_name='caller', agent=remote,
                              session_service=sessions, auto_create_session=True) as runner:
                events = [event async for event in runner.run_async(
                    user_id='u', session_id='s', new_message=types.Content(
                        role='user', parts=[types.Part(text='What is the value?')]))]
                session = await sessions.get_session(app_name='caller', user_id='u', session_id='s')
    yielded = model_texts(events)
    persisted = model_texts(session.events)
    assert yielded.count(ANSWER) == 1, yielded
    assert persisted.count(ANSWER) == 1, persisted
    assert len(calls) == int(use_tool)
    return {'executor': 'v2' if use_v2 else 'legacy',
            'scenario': 'tool' if use_tool else 'plain',
            'yielded_texts': yielded, 'persisted_texts': persisted,
            'answer_yielded': yielded.count(ANSWER),
            'answer_persisted': persisted.count(ANSWER), 'tool_calls': len(calls)}

async def main():
    if args.source:
        assert Path(executor_module.__file__).resolve().is_relative_to(args.source.resolve())
    result = {'source': executor_module.__file__,
              'python': sys.version.split()[0],
              'a2a_sdk': importlib.metadata.version('a2a-sdk'),
              'cases': [await probe(v2, tool) for v2 in (False, True) for tool in (False, True)]}
    args.output.write_text(json.dumps(result, indent=2), encoding='utf-8')
    print(json.dumps(result, indent=2))

asyncio.run(main())