[maximilliangrand]

Link to Issue or Description of Change

Fixes #4934. This targets the current A2A module, following the invitation in #4944.

When a remote task asks for input, an ordinary text or attachment reply currently sends only its context ID. A peer that requires the task ID starts another task instead of continuing the paused one. For example, replying CAD to a currency question repeats the question instead of completing the conversion.

Use the latest persisted response in the matching context and task scope to resume only INPUT_REQUIRED or AUTH_REQUIRED tasks. Persist bare pause and terminal status updates so completion cannot leave an older pause eligible for reuse. These events have content=None, preserving an earlier artifact returned through AgentTool. Bare progress updates without actions or metadata remain suppressed. Recorded request failures and incomplete-stream errors prevent reuse of an earlier pause.

Automatic task-ID inference is disabled when custom inbound converters or after-request hooks are configured: they can hide terminal responses, so their history cannot safely determine automatic continuation. Those callers retain explicit routing. Existing function-response routing, before-request hooks, context builders, and caller metadata overrides remain supported. No public API or session schema changes.

Thanks to @kthota-g for reporting this and @shriramThakare3 for the earlier contribution.

Testing Plan

• Added regression tests for text/attachments, both pause states, completion, errors, context/task isolation, session reloads, repeated pauses, reused invocation IDs, filters/hooks, and AgentTool output.
• Current upstream fails 12 selected continuation regressions; the patched cases pass.
• Full A2A suite: 816 passed, 49 skipped with a2a-sdk==1.1.5; 845 passed, 20 skipped with a2a-sdk==0.3.26.
• Pre-commit checks and wheel build pass. Mypy reports the same 733 errors as unchanged upstream, with no introduced diagnostics after normalizing line numbers.

Full Python matrix: ran the repository's locked tox environments with pytest tests/unittests -n 2 --dist loadfile. One unrelated CLI test, test_handle_login_with_google_option_3, was deliberately deselected because it starts a real interactive Google login; gcloud configuration and PATH were isolated.

Python	Passed	Failed	Skipped
3.10	16,308	12	87
3.11	16,325	0	86
3.12	16,313	3	87
3.13	16,307	9	87
3.14	16,305	11	87

Each version also reports 26 xfailed and 2 xpassed. The Python 3.10 row is a full rerun of the final commit. Broad-suite failures were checked against unchanged upstream 044a1ec3: BigQuery assertions reproduce in its full suite; LiveKit/Runner failures reproduce when their sample-loading tests run first; Python 3.12/3.14's import-allowlist tests also see the same environment-provided sitecustomize module. Every failure in the final matrix has a matching upstream reproduction. No changes to those unrelated tests or dependencies are included.

Reproduce the unrelated baseline failures

Run on unchanged upstream in the corresponding tox environment:

# The sample loader replaces the real livekit module in sys.modules.
python -m pytest 'tests/unittests/test_samples.py::test_sample_loads[integrations/livekit]' tests/unittests/integrations/livekit/test_livekit_call.py tests/unittests/integrations/livekit/test_livekit_tools.py
# The sample replay leaves its hello_world module cached.
python -m pytest 'tests/unittests/test_samples.py::test_sample[samples/core/hello_world/roll_die.json]' tests/unittests/test_runners.py::test_runner_allows_nested_agent_directories
# Same two failures on baseline and candidate with these Python 3.12/3.14 installations.
python -m pytest tests/unittests/test_import_loading.py::test_entry_point_loads_only_allowlisted_packages

The full baseline Python 3.10 run reports 8 failed, 16,226 passed, 87 skipped; all eight are identical BigQuery assertions/subtests. The original failing selection passes alone on both baseline and candidate (11 tests and 4 subtests), so isolated passing reruns were not used to dismiss the full-suite failures.

Manual end-to-end verification: installed the built wheel into a clean Python 3.12 environment with the A2A extra. A real local FastAPI JSON-RPC peer is reached through httpx.ASGITransport, using the public Runner and InMemorySessionService; no model, credentials, or external service is involved. Unchanged upstream starts two tasks and fails the assertion. The installed wheel resumes one task and returns the completed answer:

user='Convert ten USD' output=[]
user='CAD' output=['Conversion completed in CAD']
same_task= True

A second peer returns status-only INPUT_REQUIRED and COMPLETED updates. Three Runner turns resume the first task, then start a new task after completion (same_task=True, third_turn_starts_new_task=True). Unit tests cover both legacy/v2 conversion and SDK versions, including an artifact followed by bare completion.

Reproduce the JSON-RPC flow with the built wheel

uv build --wheel
uv venv --python 3.12 /tmp/adk-a2a-repro
uv pip install --python /tmp/adk-a2a-repro/bin/python 'dist/google_adk-2.10.0-py3-none-any.whl[a2a]' 'a2a-sdk[http-server]==1.1.5'
# Save the script below outside the checkout, then run it with that interpreter.
/tmp/adk-a2a-repro/bin/python /tmp/adk_a2a_repro.py

"""Credential-free Runner -> real A2A JSON-RPC -> resumable local peer probe."""

import asyncio
import warnings

import httpx
from a2a.helpers.proto_helpers import new_task, new_text_part
from a2a.server.agent_execution import AgentExecutor
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import add_a2a_routes_to_fastapi, create_agent_card_routes, create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore, TaskUpdater
from a2a.types import AgentCapabilities, AgentCard, AgentInterface, Message, Role, TaskState
from fastapi import FastAPI
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

warnings.filterwarnings('ignore')


async def main():
  observed = []

  class CurrencyPeer(AgentExecutor):

    async def execute(self, context, event_queue):
      observed.append(
          (context.get_user_input(), context.message.task_id, context.task_id)
      )
      updater = TaskUpdater(event_queue, context.task_id, context.context_id)
      if context.current_task is None:
        await event_queue.enqueue_event(
            new_task(
                context.task_id,
                context.context_id,
                TaskState.TASK_STATE_SUBMITTED,
            )
        )
        await updater.requires_input(
            Message(
                message_id='ask-currency',
                role=Role.ROLE_AGENT,
                parts=[new_text_part('Which currency?')],
            )
        )
      else:
        await updater.complete(
            Message(
                message_id='answer',
                role=Role.ROLE_AGENT,
                parts=[new_text_part('Conversion completed in CAD')],
            )
        )

    async def cancel(self, context, event_queue):
      raise NotImplementedError

  card = AgentCard(
      name='currency',
      description='currency',
      version='1',
      capabilities=AgentCapabilities(streaming=True),
      default_input_modes=['text/plain'],
      default_output_modes=['text/plain'],
      supported_interfaces=[
          AgentInterface(
              url='https://peer.test/a2a', protocol_binding='JSONRPC'
          )
      ],
  )
  peer = FastAPI()
  add_a2a_routes_to_fastapi(
      peer,
      agent_card_routes=create_agent_card_routes(card),
      jsonrpc_routes=create_jsonrpc_routes(
          DefaultRequestHandler(CurrencyPeer(), InMemoryTaskStore(), card),
          rpc_url='/a2a',
      ),
  )
  async with httpx.AsyncClient(
      transport=httpx.ASGITransport(app=peer)
  ) as client:
    agent = RemoteA2aAgent(
        name='currency', agent_card=card, httpx_client=client
    )
    runner = Runner(
        app_name='probe',
        agent=agent,
        session_service=InMemorySessionService(),
        auto_create_session=True,
    )
    try:
      for user_text in ('Convert ten USD', 'CAD'):
        output = []
        async for event in runner.run_async(
            user_id='user',
            session_id='session',
            new_message=types.Content(
                role='user', parts=[types.Part(text=user_text)]
            ),
        ):
          if event.content and event.content.parts:
            output.extend(p.text for p in event.content.parts if p.text)
        print(f'user={user_text!r} output={output!r}')
      print('peer_received=', observed)
      print('same_task=', observed[0][2] == observed[1][2])
      assert observed[1][1] == observed[0][2], observed
      assert output == ['Conversion completed in CAD'], output
    finally:
      await runner.close()


asyncio.run(main())

Checklist

• Read the contribution guide and performed self-review.
• Added regression coverage and comments for the task-state boundaries.
• Tested the built wheel end to end.
• All unit tests pass locally (unrelated baseline failures are detailed above).
• No downstream changes or documentation/API updates are required.

[adk-bot]

Thank you @maximilliangrand for your contribution! 🎉

Your changes have been successfully imported and merged via Copybara in commit 539a0d0.

Closing this PR as the changes are now in the main branch.