feat(plugins): add on_agent_error_callback and on_run_error_callback

Implements RFC #5044 to close the error callback coverage gap at the
agent and runner levels. Previously, unhandled exceptions escaping
agent execution or runner execution left dangling *_STARTING events
with no terminal error event.

Changes:
- Add on_agent_error_callback and on_run_error_callback to BasePlugin
  (notification-only, exception always re-raised)
- Wire try/except Exception in BaseAgent.run_async() and run_live()
- Wire try/except Exception in Runner._exec_with_plugin()
- Add AGENT_ERROR and INVOCATION_ERROR event types to
  BigQueryAgentAnalyticsPlugin with traceback capture and cleanup
- Keep after_agent_callback and after_run_callback as success-only
  (no semantic change to existing callbacks)
- Catch Exception (not BaseException) to exclude CancelledError/
  KeyboardInterrupt from error callback dispatch

Closes #4863

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: caohy1988

src/google/adk/agents/base_agent.py

@@ -293,9 +293,13 @@ async def run_async(
       if ctx.end_invocation:
         return
 
-      async with Aclosing(self._run_async_impl(ctx)) as agen:
-        async for event in agen:
-          yield event
+      try:
+        async with Aclosing(self._run_async_impl(ctx)) as agen:
+          async for event in agen:
+            yield event
+      except Exception as e:
+        await self._handle_agent_error_callback(ctx, e)
+        raise
 
       if ctx.end_invocation:
         return
@@ -326,9 +330,13 @@ async def run_live(
       if ctx.end_invocation:
         return
 
-      async with Aclosing(self._run_live_impl(ctx)) as agen:
-        async for event in agen:
-          yield event
+      try:
+        async with Aclosing(self._run_live_impl(ctx)) as agen:
+          async for event in agen:
+            yield event
+      except Exception as e:
+        await self._handle_agent_error_callback(ctx, e)
+        raise
 
       if event := await self._handle_after_agent_callback(ctx):
         yield event
@@ -548,6 +556,27 @@ async def _handle_after_agent_callback(
       )
     return None
 
+  async def _handle_agent_error_callback(
+      self,
+      invocation_context: InvocationContext,
+      error: Exception,
+  ) -> None:
+    """Runs the on_agent_error_callback for all plugins.
+
+    This is notification-only: the exception is always re-raised by the
+    caller after this method returns.
+
+    Args:
+      invocation_context: The invocation context for this agent.
+      error: The exception that escaped agent execution.
+    """
+    callback_context = CallbackContext(invocation_context)
+    await invocation_context.plugin_manager.run_on_agent_error_callback(
+        agent=self,
+        callback_context=callback_context,
+        error=error,
+    )
+
   @override
   def model_post_init(self, __context: Any) -> None:
     self.__set_parent_agent_for_sub_agents()

src/google/adk/plugins/base_plugin.py

@@ -370,3 +370,45 @@ async def on_tool_error_callback(
       allows the original error to be raised.
     """
     pass
+
+  async def on_agent_error_callback(
+      self,
+      *,
+      agent: BaseAgent,
+      callback_context: CallbackContext,
+      error: Exception,
+  ) -> None:
+    """Callback executed when an unhandled exception escapes agent execution.
+
+    This is a notification-only callback. The exception is always re-raised
+    after all registered plugins have been notified. Plugins should NOT
+    suppress the exception.
+
+    Unlike ``on_model_error_callback`` which can return a replacement
+    response, this callback has no return value — there is no meaningful
+    recovery action at the agent level that a plugin can provide.
+
+    Args:
+      agent: The agent instance that encountered the error.
+      callback_context: The callback context for the agent invocation.
+      error: The exception that was raised during agent execution.
+    """
+    pass
+
+  async def on_run_error_callback(
+      self,
+      *,
+      invocation_context: InvocationContext,
+      error: Exception,
+  ) -> None:
+    """Callback executed when an unhandled exception escapes runner execution.
+
+    This is a notification-only callback. The exception is always re-raised
+    after all registered plugins have been notified. Plugins should NOT
+    suppress the exception.
+
+    Args:
+      invocation_context: The context for the entire invocation.
+      error: The exception that was raised during runner execution.
+    """
+    pass

src/google/adk/plugins/bigquery_agent_analytics_plugin.py

@@ -28,6 +28,7 @@
 import logging
 import mimetypes
 import os
+import traceback as traceback_module
 
 # Enable gRPC fork support so child processes created via os.fork()
 # can safely create new gRPC channels.  Must be set before grpc's
@@ -1763,8 +1764,12 @@ def _get_events_schema() -> list[bigquery.SchemaField]:
     "AGENT_COMPLETED": [
         "CAST(JSON_VALUE(latency_ms, '$.total_ms') AS INT64) AS total_ms",
     ],
+    "AGENT_ERROR": [
+        "CAST(JSON_VALUE(latency_ms, '$.total_ms') AS INT64) AS total_ms",
+    ],
     "INVOCATION_STARTING": [],
     "INVOCATION_COMPLETED": [],
+    "INVOCATION_ERROR": [],
     "STATE_DELTA": [
         "JSON_QUERY(attributes, '$.state_delta') AS state_delta",
     ],
@@ -3279,3 +3284,108 @@ async def on_tool_error_callback(
             parent_span_id_override=None if has_ambient else parent_span_id,
         ),
     )
+
+  @_safe_callback
+  async def on_agent_error_callback(
+      self,
+      *,
+      agent: Any,
+      callback_context: CallbackContext,
+      error: Exception,
+  ) -> None:
+    """Callback when an agent execution fails with an unhandled exception.
+
+    Emits an AGENT_ERROR event and pops the agent span from TraceManager.
+    The after_agent_callback (AGENT_COMPLETED) is NOT called on failure,
+    so this callback handles the span cleanup that would otherwise be done
+    there.
+
+    Args:
+        agent: The agent instance that failed.
+        callback_context: The callback context.
+        error: The exception that escaped agent execution.
+    """
+    span_id, duration = TraceManager.pop_span()
+    parent_span_id, _ = TraceManager.get_current_span_and_parent()
+
+    has_ambient = trace.get_current_span().get_span_context().is_valid
+
+    error_tb = "".join(
+        traceback_module.format_exception(
+            type(error), error, error.__traceback__
+        )
+    )
+    # Truncate traceback to max_content_length.
+    max_len = self.config.max_content_length
+    if max_len and len(error_tb) > max_len:
+      error_tb = error_tb[:max_len] + "... [truncated]"
+
+    await self._log_event(
+        "AGENT_ERROR",
+        callback_context,
+        event_data=EventData(
+            status="ERROR",
+            error_message=str(error),
+            latency_ms=duration,
+            span_id_override=None if has_ambient else span_id,
+            parent_span_id_override=(None if has_ambient else parent_span_id),
+        ),
+        raw_content={"error_traceback": error_tb},
+    )
+
+  @_safe_callback
+  async def on_run_error_callback(
+      self,
+      *,
+      invocation_context: "InvocationContext",
+      error: Exception,
+  ) -> None:
+    """Callback when a runner execution fails with an unhandled exception.
+
+    Emits an INVOCATION_ERROR event and performs the cleanup that
+    after_run_callback (INVOCATION_COMPLETED) would normally do. Since
+    after_run_callback is success-only, this callback must handle
+    TraceManager cleanup, context var reset, and flush.
+
+    Args:
+        invocation_context: The context of the current invocation.
+        error: The exception that escaped runner execution.
+    """
+    try:
+      callback_ctx = CallbackContext(invocation_context)
+      trace_id = TraceManager.get_trace_id(callback_ctx)
+
+      span_id, duration = TraceManager.pop_span()
+      parent_span_id = TraceManager.get_current_span_id()
+
+      has_ambient = trace.get_current_span().get_span_context().is_valid
+
+      error_tb = "".join(
+          traceback_module.format_exception(
+              type(error), error, error.__traceback__
+          )
+      )
+      max_len = self.config.max_content_length
+      if max_len and len(error_tb) > max_len:
+        error_tb = error_tb[:max_len] + "... [truncated]"
+
+      await self._log_event(
+          "INVOCATION_ERROR",
+          callback_ctx,
+          event_data=EventData(
+              trace_id_override=trace_id,
+              status="ERROR",
+              error_message=str(error),
+              latency_ms=duration,
+              span_id_override=None if has_ambient else span_id,
+              parent_span_id_override=(None if has_ambient else parent_span_id),
+          ),
+          raw_content={"error_traceback": error_tb},
+      )
+    finally:
+      # Cleanup must run even if _log_event raises, otherwise
+      # stale invocation metadata leaks into the next invocation.
+      TraceManager.clear_stack()
+      _active_invocation_id_ctx.set(None)
+      _root_agent_name_ctx.set(None)
+      await self.flush()

src/google/adk/plugins/plugin_manager.py

@@ -52,6 +52,8 @@
     "after_model_callback",
     "on_tool_error_callback",
     "on_model_error_callback",
+    "on_agent_error_callback",
+    "on_run_error_callback",
 ]
 
 logger = logging.getLogger("google_adk." + __name__)
@@ -257,6 +259,34 @@ async def run_on_tool_error_callback(
         error=error,
     )
 
+  async def run_on_agent_error_callback(
+      self,
+      *,
+      agent: BaseAgent,
+      callback_context: CallbackContext,
+      error: Exception,
+  ) -> None:
+    """Runs the `on_agent_error_callback` for all plugins."""
+    await self._run_callbacks(
+        "on_agent_error_callback",
+        agent=agent,
+        callback_context=callback_context,
+        error=error,
+    )
+
+  async def run_on_run_error_callback(
+      self,
+      *,
+      invocation_context: InvocationContext,
+      error: Exception,
+  ) -> None:
+    """Runs the `on_run_error_callback` for all plugins."""
+    await self._run_callbacks(
+        "on_run_error_callback",
+        invocation_context=invocation_context,
+        error=error,
+    )
+
   async def _run_callbacks(
       self, callback_name: PluginCallbackName, **kwargs: Any
   ) -> Optional[Any]: