`A2aAgentExecutor` publishes raw exception text to the peer, bypassing the `after_agent` interceptor

[1wos]

🔴 Required Information

Describe the Bug:

When a request fails with an exception, both executors put str(e) straight into
the failure message published to the remote peer:

# src/google/adk/a2a/executor/a2a_agent_executor.py:167
# src/google/adk/a2a/executor/a2a_agent_executor_impl.py:168
parts=[_compat.make_text_part(str(e))],

The except Exception wraps the whole of _handle_request, so this covers every
failure inside the run — model calls, tool calls, database access, file I/O —
and forwards whatever those exceptions happen to carry.

The after_agent interceptor looks like the place to sanitize this, and its
docstring says it covers failed terminal events:

# src/google/adk/a2a/executor/config.py:75-79
"""Hook executed after the agent finishes and the final event is prepared.

  Allows inspection or modification of the terminal status event (e.g.,
  completed or failed) before it is enqueued. ...
"""

That holds for a failure the run produces itself: error_event becomes
final_event and passes through execute_after_agent_interceptors at
a2a_agent_executor_impl.py:249.

It does not hold for the exception path. The except block builds its own event
and enqueues it at line 172, after the exception has already abandoned
_handle_request, so line 249 never runs. An operator who registers after_agent
to redact terminal messages gets redaction on the first path and a verbatim leak
on the exception path.

Steps to Reproduce:

1. Serve an agent behind A2aAgentExecutor
2. Make the run raise with an infrastructure detail in the message (a missing
   credentials file, a DB connection string, a rejected API key)
3. Read the failed task status message the peer receives
4. Register an after_agent interceptor that rewrites the message and observe
   that it is not consulted on this path

Expected Behavior:

The peer learns the request failed. Operator-facing detail stays in the server's
logs, and if after_agent is documented to cover failed terminal events, it is
consulted for this one too.

Observed Behavior:

The exception message is delivered verbatim. Driving the existing failure path
with three exceptions of a kind a real run raises:

REMOTE PEER RECEIVES: "[Errno 2] No such file or directory:
                       '/Users/me/.config/gcloud/service-account-key.json'"
REMOTE PEER RECEIVES: 'connection failed: postgres://svc_agent:REDACTED@10.0.0.7:5432/prod'
REMOTE PEER RECEIVES: 'Incorrect API key provided: sk-proj-AbCd1234...Zz. '

A peer that only sent the agent a question receives the server's filesystem
layout, an internal host and port, a database password and part of an API key.

Environment Details:

• ADK Library Version: 2.10.0 (reproduced on main @ 044a1ec)
• Desktop OS: macOS
• Python Version: 3.12.12

Model Information:

• Are you using LiteLLM: No
• Which model is being used: N/A — the failure path does not reach a model.

---

🟡 Optional Information

Regression: N/A — not checked against earlier versions.

Minimal Reproduction Code:

The repo's own failure-path test already drives this; it asserts the task state
but not the message text. Extending it shows what is forwarded:

# based on tests/unittests/a2a/executor/test_a2a_agent_executor.py
# (test_execute_with_exception, which already exercises this path)
self.mock_context.task_id = "t1"
self.mock_context.current_task = None
self.mock_request_converter.side_effect = FileNotFoundError(
    2, "No such file or directory", "/Users/me/.config/gcloud/sa-key.json")

await self.executor.execute(self.mock_context, self.mock_event_queue)

event = self.mock_event_queue.enqueue_event.call_args_list[-1][0][0]
print(event.status.message.parts[0].text)
# [Errno 2] No such file or directory: '/Users/me/.config/gcloud/sa-key.json'

How often has this issue occurred?: Always (100%)

Additional Context:

Both executors are affected, so a fix needs a2a_agent_executor.py:167 and
a2a_agent_executor_impl.py:168.

A fix that keeps failures debuggable:

1. Log the exception in full against a short generated id.
2. Send the peer a fixed summary carrying only that id.
3. Put the exception text behind an explicit opt-in for local debugging.

Routing this event through execute_after_agent_interceptors as well would make
the after_agent docstring true for both failure paths.

One existing assertion would need relaxing:
tests/unittests/a2a/executor/test_a2a_agent_executor_impl.py checks the failure
message text.

I am happy to send a PR if this is accepted as a bug.

[sattyamjjain]

+1 on #7317. One thing that makes it low risk, the same except block already logs the full exception server-side with exc_info=True, so sending a generic message to the peer doesn't lose anything for the operator:

adk-python/src/google/adk/a2a/executor/a2a_agent_executor.py

Lines 153 to 172 in def458b

	await self._handle_request(context, event_queue)
	except Exception as e:
	logger.error('Error handling A2A request: %s', e, exc_info=True)
	# Publish failure event
	try:
	await event_queue.enqueue_event(
	_compat.make_task_status_update_event(
	task_id=context.task_id,
	context_id=context.context_id,
	status=_compat.make_task_status(
	_compat.TS_FAILED,
	message=Message(
	message_id=platform_uuid.new_uuid(),
	role=_compat.ROLE_AGENT,
	parts=[_compat.make_text_part(str(e))],
	),
	),
	final=True,
	)
	)

[llalitkumarrr]

Hello @GWeale,

Could you please take a look at this? We are able to reproduce the issue in version 2.10.0. We also took a look the linked PR, it seems to partially fix the problem. We would appreciate your input on this.