A2A: 2.7.0 flattens the answer to a RELAYED human-input pause, so a remote agent can never resume (silent approval-gate bypass)

## 🔴 Required Information

**Describe the Bug:**

`RemoteA2aAgent` in 2.7.0 flattens a human-input function response to text before
forwarding it (`aec7aa3`, "fix(a2a): flatten human-input responses on resume to
avoid mixing them with text"). The rewrite decides what to flatten by **function
call name**:

```python
_HUMAN_INPUT_FUNCTION_CALL_NAMES = frozenset({
MOCK_FUNCTION_CALL_FOR_REQUIRED_USER_INPUT,
MOCK_FUNCTION_CALL_FOR_REQUIRED_USER_AUTH,
REQUEST_INPUT_FUNCTION_CALL_NAME,
REQUEST_CONFIRMATION_FUNCTION_CALL_NAME,
REQUEST_EUC_FUNCTION_CALL_NAME,
})
```

That is correct for a pause the **caller** raised — the remote agent never saw
that call, so a `FunctionResponse` addressed to it would be meaningless. But a
pause raised **inside the remote agent** and relayed up to the caller carries the
same name. ADK creates exactly that shape itself: when a remote task goes
`input-required`, the genuine long-running call is re-emitted into the caller's
session (`a2a/converters/to_adk_event.py:160-171`), which is what makes a nested
human-in-the-loop pause visible to the human at the top.

When the human answers, the response is flattened, so the remote agent never
receives a `FunctionResponse` for the call it is still paused on. It cannot
resume. Worse, under the legacy executor the text is accepted as a *new user
turn*, so the run continues and produces a confident answer while the gated tool
never executes.

**Steps to Reproduce:**

1. `pip install "google-adk[a2a]==2.7.0" "a2a-sdk[http-server]==0.3.26"`
2. Save the script under "Minimal Reproduction Code" as `check_resume.py`. It
builds the exact session shape ADK produces for a relayed pause — the peer's
long-running `adk_request_confirmation` call, then the human's
`FunctionResponse` — and asks `RemoteA2aAgent` what it would send.
3. `python check_resume.py`

**Expected Behavior:**

The answer to a pause that was raised *by the remote agent* is forwarded to it as
a function response, so it resumes the invocation it is blocked on. This is what
2.6.1 did.

**Observed Behavior:**

```
outbound message parts:
TextPart: '{"confirmed": true, "hint": "Approve publish_result()?", "payload": {"note": "approved"}}'
```

The `FunctionResponse` is gone. Consequences, in ascending severity:

* The remote agent's paused invocation is never resumed.
* With the newer executor, the request is rejected with *"It was not provided a
function response for the function call"*
(`a2a/converters/long_running_functions.py:192`) — at least it is loud.
* With the legacy executor it is **silent**. See the end-to-end run below: the
human approved, the approval-gated tool never executed, and the caller was
handed a fabricated tool result.

**Environment Details:**

- ADK Library Version: 2.7.0 (works on 2.6.1)
- Desktop OS: macOS (Apple Silicon)
- Python Version: 3.14.6
- a2a-sdk: 0.3.26

**Model Information:**

- Are you using LiteLLM: No
- Which model is being used: gemini-2.5-pro (caller), gemini-2.5-flash
(intermediate), gemini-2.5-flash-lite (leaf). The bug is in message
construction and does not depend on the model.

---

## 🟡 Optional Information

**Regression:**

Yes — 2.6.1 forwards the response as a `DataPart` with
`adk_type: function_response` and the nested resume works. Introduced by
`aec7aa3` in 2.7.0.

**Logs:**

Three agents chained over A2A, `orchestrator -> research -> math`, with an
approval-gated tool (`require_confirmation`) on the leaf. The leaf logs a marker
line when the gated tool body actually runs.

The pause propagates correctly and reaches the human:

```text
[math] EVENT#4 author=math lrt=['adk-d49377f6-…'] parts=['fc(adk_request_confirmation)']
[math] OUT status state=input_required final=True parts=['DATA(name=adk_request_confirmation)']
[research] IN from=math state=input_required
[research] OUT status state=input_required final=True parts=['DATA(name=adk_request_confirmation)']
→ human sees the approval request
```

The human approves. The leaf receives **no further request at all** — its log
ends at the pause above — and the intermediate agent answers from the flattened
text instead:

```text
[research] EVENT#1 author=research parts=['text[95]\'The math tool returned:
"FunctionResponse(tool_name=\'publish_result\', content={\'result\': 437})"\'']
[research] OUT artifact last_chunk=True author=research parts=[… same text …]
```

Count of the leaf's "gated tool executed" marker for this run: **0**. The tool
result in that sentence was invented by the intermediate agent's model from the
JSON text it was handed. The caller reported the turn as successfully resumed.

With the response kept as a function response, the same chain resumes correctly
and the gated tool runs — 5/5 consecutive runs.

**Additional Context:**

A possible fix is to classify by **origin** rather than by call name: do not
flatten when the matching function-call event came from the remote agent itself.
`RemoteA2aAgent._is_remote_response(event)` already answers that question, and
the relayed pause event carries the A2A task metadata
(`a2a:task_id`) that `_create_a2a_request_for_user_function_response` reads a few
lines later. Locally raised pauses would keep the new behaviour unchanged.

We are unblocked by overriding
`_create_a2a_request_for_user_function_response` in a `RemoteA2aAgent` subclass:
if the matching call event is a remote response from that peer and every function
response in the answer is `adk_request_confirmation` / `adk_request_input`, we
send the function-response parts as data. Credential and auth responses are left
to ADK's new behaviour, so the security intent of `aec7aa3` is preserved.

**Minimal Reproduction Code:**

```python
"""ADK 2.7.0: the answer to a RELAYED pause is flattened and stops being a resume."""

from types import SimpleNamespace

from google.adk.agents.remote_a2a_agent import RemoteA2aAgent
from google.adk.events.event import Event
from google.genai import types

CALL_ID = "adk-abc123"

# 1. The remote agent's long-running pause, as ADK relays it into the caller's
# session (a2a/converters/to_adk_event.py:143).
relayed_pause = Event(
author="research",
invocation_id="inv-1",
long_running_tool_ids={CALL_ID},
content=types.Content(
role="model",
parts=[
types.Part(
function_call=types.FunctionCall(
id=CALL_ID,
name="adk_request_confirmation",
args={
"originalFunctionCall": {
"id": "p1",
"name": "publish_result",
"args": {"value": "437.0"},
},
"toolConfirmation": {"hint": "Approve publish_result()?"},
},
)
)
],
),
)

# 2. The human's answer to it.
human_answer = Event(
author="user",
invocation_id="inv-1",
content=types.Content(
role="user",
parts=[
types.Part(
function_response=types.FunctionResponse(
id=CALL_ID,
name="adk_request_confirmation",
response={
"confirmed": True,
"hint": "Approve publish_result()?",
"payload": {"note": "approved"},
},
)
)
],
),
)

ctx = SimpleNamespace(
session=SimpleNamespace(
id="s1", app_name="app", user_id="u", events=[relayed_pause, human_answer]
),
app_name="app",
user_id="u",
invocation_id="inv-1",
branch=None,
)

peer = RemoteA2aAgent(name="research", agent_card="http://127.0.0.1:8091/card.json")
message = peer._create_a2a_request_for_user_function_response(ctx)

print("outbound message parts:")
for part in message.parts:
root = part.root
if type(root).__name__ == "TextPart":
print(f" TextPart: {root.text!r}")
else:
print(f" {type(root).__name__}: {root.data} meta={root.metadata}")

# 2.6.1: DataPart with adk_type=function_response -> the peer resumes.
# 2.7.0: TextPart -> the peer starts a new turn.
```

**How often has this issue occurred?:**

- Always (100%)