RemoteA2aAgent: streamed A2A chunks and the final event that replaces them get different event ids, so the answer can be shown twice #7395

# Partial chunks of one streaming response share the base id; mint a
# fresh id only after a complete event so distinct responses differ.
if not event.partial:
  model_response_event.id = Event.new_id()

import asyncio, socket, threading
from typing import AsyncGenerator

import httpx, uvicorn
from a2a.client.client import ClientConfig
from a2a.client.client_factory import ClientFactory
from google.adk.a2a.converters.request_converter import convert_a2a_request_to_agent_run_request
from google.adk.a2a.executor.a2a_agent_executor import A2aAgentExecutor
from google.adk.a2a.executor.config import A2aAgentExecutorConfig
from google.adk.a2a.utils.agent_to_a2a import to_a2a
from google.adk.agents.llm_agent import LlmAgent
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent
from google.adk.agents.run_config import RunConfig, StreamingMode
from google.adk.models.base_llm 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.in_memory_session_service import InMemorySessionService
from google.genai import types

CHUNKS = ["Hello, ", "this is ", "a streamed answer."]


class StreamingFakeLlm(BaseLlm):
  model: str = "fake-streaming"

  async def generate_content_async(self, llm_request: LlmRequest, stream: bool = False) -> AsyncGenerator[LlmResponse, None]:
    if stream:
      for text in CHUNKS:
        yield LlmResponse(content=types.Content(role="model", parts=[types.Part(text=text)]), partial=True)
    yield LlmResponse(content=types.Content(role="model", parts=[types.Part(text="".join(CHUNKS))]), partial=False, turn_complete=True)


def sse_request(request, part_converter=None):
  run_request = convert_a2a_request_to_agent_run_request(request, part_converter) if part_converter else convert_a2a_request_to_agent_run_request(request)
  run_request.run_config = RunConfig(streaming_mode=StreamingMode.SSE)
  return run_request


def serve(port: int) -> None:
  agent = LlmAgent(name="streamer", model=StreamingFakeLlm(), instruction="Answer.")
  app = to_a2a(agent, host="127.0.0.1", port=port, agent_executor_factory=lambda runner: A2aAgentExecutor(
      runner=runner, config=A2aAgentExecutorConfig(request_converter=sse_request), force_new_version=True))
  uvicorn.run(app, host="127.0.0.1", port=port, log_level="error")


async def main() -> None:
  with socket.socket() as s:
    s.bind(("127.0.0.1", 0)); port = s.getsockname()[1]
  threading.Thread(target=serve, args=(port,), daemon=True).start()
  base = f"http://127.0.0.1:{port}"
  async with httpx.AsyncClient() as probe:
    for _ in range(100):
      try:
        if (await probe.get(f"{base}/.well-known/agent-card.json")).status_code == 200:
          break
      except httpx.HTTPError:
        pass
      await asyncio.sleep(0.1)
  client = RemoteA2aAgent(name="streamer", agent_card=f"{base}/.well-known/agent-card.json", use_legacy=False,
      a2a_client_factory=ClientFactory(config=ClientConfig(streaming=True, httpx_client=httpx.AsyncClient(timeout=30))))
  runner = Runner(app_name="e2e", agent=client, session_service=InMemorySessionService())
  session = await runner.session_service.create_session(app_name="e2e", user_id="u")
  ids = set()
  async for event in runner.run_async(user_id="u", session_id=session.id, new_message=types.Content(role="user", parts=[types.Part(text="Hi")])):
    text = "".join(p.text or "" for p in (event.content.parts if event.content else []) if not p.thought)
    if text:
      ids.add(event.id)
      print(f"partial={str(event.partial):5}  id={event.id[:13]}…  text={text!r}")
  print(f"distinct ids: {len(ids)}")


asyncio.run(main())