fix(plugins): make notification dispatch best-effort on callback failure

_run_notification_callbacks() previously raised RuntimeError on the
first plugin callback failure, which masked the original application
exception and prevented later plugins from being notified.

Now it logs the error and continues iterating all plugins. The caller's
original exception (agent crash / runner crash) remains the primary
error that propagates to the caller.

Add regression tests:
- test_plugin_callback_failure_does_not_mask_app_error: verifies p2
  is still notified when p1's error callback raises
- test_original_exception_propagates_despite_plugin_failure: end-to-end
  test verifying the caller sees the original agent exception, not the
  plugin's internal error

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

Author: caohy1988

src/google/adk/plugins/plugin_manager.py

@@ -341,29 +341,29 @@ async def _run_notification_callbacks(
   ) -> None:
     """Executes a notification-only callback for all registered plugins.
 
-    Unlike ``_run_callbacks``, this method **always iterates all plugins**
-    regardless of return values.  It is intended for error callbacks where
-    every plugin must be notified (the return value is discarded).
+    Unlike ``_run_callbacks``, this method is **best-effort**: it always
+    iterates all plugins regardless of return values or exceptions.  If a
+    plugin's callback raises, the error is logged and iteration continues
+    so that every plugin gets notified.  This is critical for error
+    callbacks where the caller holds an original application exception
+    that must remain the primary error.
 
     Args:
       callback_name: The name of the callback method to execute.
       **kwargs: Keyword arguments to be passed to the callback method.
-
-    Raises:
-      RuntimeError: If a plugin encounters an unhandled exception during
-        execution. The original exception is chained.
     """
     for plugin in self.plugins:
       callback_method = getattr(plugin, callback_name)
       try:
         await callback_method(**kwargs)
       except Exception as e:
-        error_message = (
-            f"Error in plugin '{plugin.name}' during '{callback_name}'"
-            f" callback: {e}"
+        logger.error(
+            "Error in plugin '%s' during '%s' callback: %s",
+            plugin.name,
+            callback_name,
+            e,
+            exc_info=True,
         )
-        logger.error(error_message, exc_info=True)
-        raise RuntimeError(error_message) from e
 
   async def close(self) -> None:
     """Calls the close method on all registered plugins concurrently.

tests/unittests/plugins/test_error_callbacks.py

@@ -500,3 +500,72 @@ async def on_run_error_callback(self, **kwargs):
     # Both plugins must be called even though p1 returns non-None.
     assert p1.run_error_called
     assert p2.run_error_called
+
+  @pytest.mark.asyncio
+  async def test_plugin_callback_failure_does_not_mask_app_error(self):
+    """When a plugin's error callback raises, iteration continues
+    and the original application exception is what the caller sees."""
+
+    class _FailingPlugin(BasePlugin):
+      __test__ = False
+
+      def __init__(self, name):
+        super().__init__(name)
+        self.agent_error_called = False
+        self.run_error_called = False
+
+      async def on_agent_error_callback(self, **kwargs):
+        self.agent_error_called = True
+        raise ValueError("plugin boom")
+
+      async def on_run_error_callback(self, **kwargs):
+        self.run_error_called = True
+        raise ValueError("plugin boom")
+
+    p1 = _FailingPlugin(name="p1")
+    p2 = _ErrorTrackingPlugin(name="p2")
+    pm = PluginManager(plugins=[p1, p2])
+
+    # Agent error callback: p1 raises, p2 must still be notified.
+    mock_agent = Mock(spec=BaseAgent)
+    mock_agent.name = "test_agent"
+    await pm.run_on_agent_error_callback(
+        agent=mock_agent,
+        callback_context=Mock(spec=CallbackContext),
+        error=RuntimeError("app crash"),
+    )
+    assert p1.agent_error_called
+    assert len(p2.agent_errors) == 1
+
+    # Run error callback: same behavior.
+    await pm.run_on_run_error_callback(
+        invocation_context=Mock(spec=InvocationContext),
+        error=RuntimeError("app crash"),
+    )
+    assert p1.run_error_called
+    assert len(p2.run_errors) == 1
+
+  @pytest.mark.asyncio
+  async def test_original_exception_propagates_despite_plugin_failure(
+      self,
+  ):
+    """End-to-end: a crashing plugin error callback does not mask
+    the original agent exception seen by the caller."""
+
+    class _FailingPlugin(BasePlugin):
+      __test__ = False
+
+      def __init__(self, name):
+        super().__init__(name)
+
+      async def on_agent_error_callback(self, **kwargs):
+        raise ValueError("plugin internal error")
+
+    plugin = _FailingPlugin(name="bad_plugin")
+    agent = _CrashingAgent(name="crash_agent")
+    ctx = await _create_ctx(agent, plugins=[plugin])
+
+    # The caller must see the original RuntimeError("agent crashed"),
+    # NOT the plugin's ValueError.
+    with pytest.raises(RuntimeError, match="agent crashed"):
+      _ = [e async for e in agent.run_async(ctx)]