feat(plugin): implement on_pipeline_error_callback lifecycle hook

[mramansayyad]

Description

This PR implements the on_pipeline_error_callback lifecycle hook on BasePlugin and PluginManager, bringing adk-python to feature parity with the corresponding capabilities in adk-go.

Associated Issue

This PR introduces the feature/parity capability directly without a separate GitHub issue, following the issue template structure below:

• Feature/Problem: Currently, adk-python does not expose a lifecycle hook to capture pipeline-level/runner-level errors across plugins. In contrast, adk-go provides on_pipeline_error_callback, which allows plugins to clean up resources, rollback operations, or report/format errors before they propagate out.
• Proposed Solution:
  1. Add the abstract/no-op async method on_pipeline_error_callback(self, error: Exception, **kwargs) -> Exception to BasePlugin.
  2. Implement run_on_pipeline_error_callback(self, invocation_context: InvocationContext, error: Exception) -> Exception in PluginManager to chain the error callbacks across registered plugins.
  3. Integrate the hook into the core execution runners (runners.py) by wrapping the primary yield-loops in try...except blocks that invoke the pipeline error hook before re-raising or modifying the exception.

---

Logs or Screenshots

Executing the unit tests verifies the correctness of the lifecycle hooks:

============================= test session starts =============================
platform win32 -- Python 3.10.11, pytest-9.0.3, pluggy-1.6.0
rootdir: D:\Projects\GSSOC\adk-python
configfile: pyproject.toml
plugins: anyio-4.12.1, asyncio-1.4.0, mock-3.15.1
collected 14 items

tests\unittests\plugins\test_plugin_manager.py ..............            [100%]
======================= 14 passed, 4 warnings in 2.99s ========================

---

Testing Plan

A dedicated set of unit tests was added to tests/unittests/plugins/test_plugin_manager.py:

1. test_all_callbacks_are_supported: Verifies on_pipeline_error_callback is registered and callable on PluginManager.
2. test_pipeline_error_callback_chaining: Verifies that the exception is passed through the registered plugins sequentially and can be modified/returned.
3. test_pipeline_error_callback_exception_wrap: Verifies that plugin callback failures are cleanly wrapped in a RuntimeError and propagated.

How to Run the Tests

To verify locally, run the following command in the adk-python directory:

pytest tests/unittests/plugins/test_plugin_manager.py

---

Type of change

• Bug fix (non-breaking change that fixes an issue)
• New feature (non-breaking change that adds functionality)
• Breaking change (alters existing behavior or a public API)
• Docs / tests / internal tooling (no user-facing behavior change)

Checklist

• I have read the contribution guidelines.
• Tests cover this change and pass locally.
• Public APIs and non-trivial functions are documented.
• If this change is user-facing, I have updated CHANGELOG.md

[adk-bot]

Response from ADK Triaging Agent

Hello @mramansayyad, thank you for submitting this pull request!

To help us review your contribution more efficiently, please ensure that it aligns with our contribution guidelines. Currently, the following items are missing from your PR description:

1. Associated Issue: All PRs, except for small documentation/typo fixes, should have an associated issue. If a relevant issue doesn't exist, please create one first, or describe the feature directly following our issue template structure in the PR body.
2. Logs or Screenshots: For new features, please provide logs or a screenshot demonstrating the changes or outputs after the implementation is applied.
3. Testing Plan: Please include a dedicated testing plan section in your PR description outlining how you verified this change (including instructions on how to run any tests you added).

Providing this context will save time for the reviewers. Thank you for your cooperation!

[rohityan]

/adk-pr-analyze

[adk-bot]

🔍 ADK Pull Request Analysis: PR #5988

Title: feat(plugin): implement on_pipeline_error_callback lifecycle hook
Author: @mramansayyad
Status: OPEN
Impact: 209 additions, 104 deletions across 4 files

Executive Summary

1. Core Objective: Implement the on_pipeline_error_callback lifecycle hook on BasePlugin and PluginManager, and integrate it within the runner pipeline wrapper _exec_with_plugin in runners.py to globally capture and transform pipeline execution errors across all plugins, matching the feature set of adk-go.
2. Justification & Value: Valuable Feature - Fills a critical gap in execution observability where plugins previously could only intercept errors at the model and tool levels rather than globally tracking the complete runner pipeline execution failures.
3. Alignment with Principles: Pass with Nits - Outstanding adherence to keyword-only arguments, type annotations, and Pydantic conventions, with minor PEP 8 spacing nits inside the test file.
4. Recommendation: Approve with Nits (Requesting clean-up of minor PEP 8 blank line formatting nits in the test suite).

Detailed Findings & Analysis

1. Objectives & Impact ("What does it do?")

• Context & Background: Currently, standard plugins only support error interceptors at the model level via on_model_error_callback and tool level via on_tool_error_callback. Global runner failures (such as graph layout crashes, database errors, or uncaught exceptions) could only be monitored from the master caller, limiting the logging or telemetry capability of individual plugins. This PR provides a parity implementation for global pipeline/runner error capture.
• Implementation Mechanism:
  • base_plugin.py: Defines a fallback on_pipeline_error_callback method returning the exception unchanged.
  • plugin_manager.py: Enhances PluginCallbackName literal definition and adds run_on_pipeline_error_callback to sequentially chain and process errors across registered plugins, lifting plugin crashes as standard RuntimeError.
  • runners.py: Wraps the pipeline execution loop in standard try...except...finally structure, catching any pipeline exceptions and invoking the new hook sequentially.
  • test_plugin_manager.py: Implements mock plugin verification, exception wrapping, and chaining regression coverage.
• Affected Surface: All changes are fully backwards-compatible. Preexisting third-party plugins extending BasePlugin do not require code changes and will inherit the default pass/return implementation.

---

2. Justification & Value ("Is it a valid and useful change?")

• Workspace Verification:
  • Investigated base_plugin.py and plugin_manager.py and confirmed that there was no existing runner/pipeline level error intercept mechanism in the baseline workspace.
• Value Assessment: High. Adding hook capability for telemetry/observability plugins to monitor complete runner pipeline results enables advanced developer logging, retry-mechanisms, error notification reporting, and profiling across plugins.
• Alternative Approaches: Sequential execution that transforms and bubbles the error iteratively is the most elegant model, mirroring adk-go.
• Scope & Depth: Systematic Fix / Root Cause.

---

3. Principle & Style Alignment Checklist ("Does it follow rules?")

• Public API & Visibility Boundaries:
  • Status: Pass
  • Analysis: No breaking shifts are introduced. The added instance method uses keyword-only arguments cleanly via * format matching ADK conventions:

  async def on_pipeline_error_callback(
      self,
      *,
      invocation_context: InvocationContext,
      error: Exception,
  ) -> Exception:

• Code Quality, Typing & Conventions:
  • Status: Nits
  • Analysis: Proper typing references, annotations, and structured exceptions are maintained. However, the PR introduces PEP 8 spacing violations within test_plugin_manager.py:
  • Inserts a third blank line between test_all_callbacks_are_supported and @pytest.mark.asyncio.
  • Inserts a third blank line before test_pipeline_error_callback_chaining.
  • Introduces an unnecessary trailing blank line at the end of the file.
    These PEP 8 violations causes the automated check run pre-commit to complete with a FAILURE.
• Robustness & Edge Cases:
  • Status: Pass
  • Analysis: Error catch inside run_on_pipeline_error_callback is carefully protected and wraps potential plugin failures in a traceback-friendly RuntimeError wrapper.
• Test Integrity & Quality:
  • Status: Pass
  • Analysis: Outstanding testing coverage is provided by test_pipeline_error_callback_chaining and test_pipeline_error_callback_exception_wrap verifying both normal error mapping chaining and exception isolation.

[adk-bot]

🔍 ADK Pull Request Analysis: PR #5988

Title: feat(plugin): implement on_pipeline_error_callback lifecycle hook
Author: @mramansayyad
Status: OPEN
Impact: 204 additions, 105 deletions across 4 files

Executive Summary

1. Core Objective: Implement the on_pipeline_error_callback lifecycle hook on base_plugin.py and plugin_manager.py, integrating it sequentially within the runner execution wrapper _exec_with_plugin in runners.py to enable global exception intercepting, transforming, and tracking across plugins, achieving parity with adk-go.
2. Justification & Value: Valuable Feature - Fills a major execution lifecycle observability gap. Telemetry and logging plugins previously could only capture localized errors at the model and tool levels rather than globally monitoring complete pipeline execution failures.
3. Alignment with Principles: Pass with Nits - Exceptionally well-aligned with key-only arguments, type assertions, and Pydantic conventions. There is a minor Lazy Logging violation in the exception handler of plugin_manager.py.
4. Recommendation: Approve with Nits (Requesting clean-up of the formatting logging nit).

Detailed Findings & Analysis

1. Objectives & Impact ("What does it do?")

• Context & Background: Existing logger and interceptor plugins could only monitor exceptions raised during specific steps (model endpoints, tool calls). Any master/runner wrapper failures (such as graph definition crashes or overall execution issues) could not be captured inside standard plugins. This PR introduces a global pipeline/runner level callback hook.
• Implementation Mechanism:
  • base_plugin.py: Declares the new on_pipeline_error_callback(...) lifecycle callback on the abstract BasePlugin class, returning the original exception unmodified by default.
  • plugin_manager.py: Adds "on_pipeline_error_callback" list literals, and exposes run_on_pipeline_error_callback which loops through all registered plugins and passes the mutated error sequentially. Any plugin-level crash is isolated and wrapped into a standard RuntimeError.
  • runners.py: Wraps the entire runner execution flow inside _exec_with_plugin under a standard try...except Exception as e: block. If an execution exception occurs, it passes it through the new plugin hook before raising it to the master caller.
  • test_plugin_manager.py: Extends mock plugins and provides robust coverage for callback registers, chaining mutations, and nested error handlers.
• Affected Surface: All additions are completely backwards-compatible; no current custom plugins subclassing BasePlugin will break.

---

2. Justification & Value ("Is it a valid and useful change?")

• Workspace Verification:
  • Investigated base_plugin.py and plugin_manager.py in the workspace and verified that there was no existing global pipeline-level callback hook to handle global execution failures.
• Value Assessment: High. Adding hook-interception capabilities at the pipeline execution level allows for extensive developer telemetry, standardized error aggregation, custom notification/alert setups, and profiling directly built inside plugin libraries.
• Alternative Approaches: Chaining error values sequentially matches the paradigm of standard ADK-Go plugin integrations and represents the cleanest design choice to allow subsequent plugins to see transformed/custom exceptions.
• Scope & Depth: Systematic Fix / Root Cause.

---

3. Principle & Style Alignment Checklist ("Does it follow rules?")

• Public API & Visibility Boundaries:
  • Status: Pass
  • Analysis: Fully non-breaking. The added signature uses *, correctly to enforce keyword-only arguments cleanly:

  async def on_pipeline_error_callback(
      self,
      *,
      invocation_context: InvocationContext,
      error: Exception,
  ) -> Exception:

• Code Quality, Typing & Conventions:
  • Status: Nits
  • Analysis: Outstanding adherence to typing structures.
  • Style Nit - Lazy Logging:
    In plugin_manager.py:L288-294:

    except Exception as e:
      error_message = (
          f"Error in plugin '{plugin.name}' during "
          f"'on_pipeline_error_callback' callback: {e}"
      )
      logger.error(error_message, exc_info=True)
      raise RuntimeError(error_message) from e

    The logging of error_message uses eager f-string evaluation which violates the ADK guidelines for Lazy Logging. It should be changed to use a lazy %-based template:

    except Exception as e:
      error_message = (
          f"Error in plugin '{plugin.name}' during "
          f"'on_pipeline_error_callback' callback: {e}"
      )
      logger.error(
          "Error in plugin '%s' during 'on_pipeline_error_callback' callback: %s",
          plugin.name,
          e,
          exc_info=True,
      )
      raise RuntimeError(error_message) from e

• Robustness & Edge Cases:
  • Status: Pass
  • Analysis: If plugin_manager is None (for executions without loaded plugins), runners.py:L1479-1485 safely bypasses callback calls and raises the original exception.
• Test Integrity & Quality:
  • Status: Pass
  • Analysis: Extensive coverage. test_pipeline_error_callback_chaining covers chaining verification, and test_pipeline_error_callback_exception_wrap handles isolated plugin faults gracefully.

Currently, the active local branch has been safely returned to main.

[mramansayyad]

Hello, I have updated the PR description to include the associated issue context, testing plan, and local test logs. Please let me know if any other changes are needed! Thank you.

[mramansayyad]

/adk-pr-analyze