From 08a20fe2d75c55e9816be6c6865961425bfffe7e Mon Sep 17 00:00:00 2001 From: Guan-Ming Chiu <105915352+guan404ming@users.noreply.github.com> Date: Thu, 17 Sep 2026 17:34:34 +0800 Subject: [PATCH] Support notifiers in LLM approval reviews (#72159) * Support notifiers in LLM approval reviews A pending LLM approval review was only visible on the Required Actions page, so nobody was told a task was waiting and it sat until someone happened to look. * Pin notifier template errors under strict Dag rendering --- providers/common/ai/docs/operators/llm.rst | 18 ++++ .../common/ai/docs/operators/llm_branch.rst | 4 +- .../ai/docs/operators/llm_file_analysis.rst | 4 +- .../ai/docs/operators/llm_schema_compare.rst | 6 +- .../common/ai/docs/operators/llm_sql.rst | 3 + .../common/ai/example_dags/example_llm.py | 13 +++ .../providers/common/ai/mixins/approval.py | 26 ++++++ .../providers/common/ai/operators/llm.py | 22 ++++- .../common/ai/operators/llm_branch.py | 2 +- .../common/ai/operators/llm_file_analysis.py | 2 +- .../common/ai/operators/llm_schema_compare.py | 2 +- .../providers/common/ai/operators/llm_sql.py | 2 +- .../unit/common/ai/mixins/test_approval.py | 91 ++++++++++++++++++- .../unit/common/ai/operators/test_llm.py | 27 ++++++ 14 files changed, 211 insertions(+), 11 deletions(-) diff --git a/providers/common/ai/docs/operators/llm.rst b/providers/common/ai/docs/operators/llm.rst index 50f8ae56216..ecbbfe3584a 100644 --- a/providers/common/ai/docs/operators/llm.rst +++ b/providers/common/ai/docs/operators/llm.rst @@ -254,6 +254,22 @@ Reject the primary button: :start-after: [START howto_operator_llm_approval] :end-before: [END howto_operator_llm_approval] +A pending review is not surfaced as a notification. Pass +``approval_notifiers`` to tell the reviewers about it through any Airflow +notifier (Slack, email, ...), the way +:class:`~airflow.providers.standard.operators.hitl.HITLOperator` does with +``notifiers``. The notifiers run once the review is open and can reference +the review ``{{ task.subject }}`` and ``{{ task.body }}`` in their templates, +as the example above does. The ``@task.llm`` decorator and the operator +subclasses accept the same parameter. A notifier whose delivery fails is +logged and the task still waits for the review; a template error fails the +task. A retry re-runs the LLM and re-notifies with the regenerated output, +while the open review keeps the original subject and body. + +The default ``body`` contains the rendered prompt and the output. Where either +is sensitive, template only ``{{ task.subject }}`` and a link to the review +into channels outside Airflow's auth boundary. + Parameters ---------- @@ -279,6 +295,8 @@ Parameters ``require_approval=True`` and a positive ``approval_timeout``. - ``allow_modifications``: If ``True``, the reviewer can edit the output before approving. Default ``False``. +- ``approval_notifiers``: Notifier, or list of notifiers, called once the review + is open. Default ``None``. Logging ------- diff --git a/providers/common/ai/docs/operators/llm_branch.rst b/providers/common/ai/docs/operators/llm_branch.rst index eba21bd3e55..c8999bfa003 100644 --- a/providers/common/ai/docs/operators/llm_branch.rst +++ b/providers/common/ai/docs/operators/llm_branch.rst @@ -111,7 +111,7 @@ call. Apart from ``fail_on_reject`` and ``ignore_downstream_trigger_rules``, which are specific to this operator, ``approval_timeout``, ``on_approval_timeout``, -and the rest of the approval behaviour are inherited from +``approval_notifiers``, and the rest of the approval behaviour are inherited from :ref:`LLMOperator `. How It Works @@ -147,6 +147,8 @@ Parameters ``require_approval=True`` and a positive ``approval_timeout``. - ``allow_modifications``: If ``True``, the reviewer can change the chosen branch(es) before approving. Default ``False``. +- ``approval_notifiers``: Notifier, or list of notifiers, called once the review + is open. Default ``None``. - ``fail_on_reject``: If ``True``, a rejected review fails the task instead of skipping the downstream tasks. Generally discouraged. Only takes effect with ``require_approval=True``. Default ``False``. diff --git a/providers/common/ai/docs/operators/llm_file_analysis.rst b/providers/common/ai/docs/operators/llm_file_analysis.rst index dc65e31bde0..fc9ce6ac29f 100644 --- a/providers/common/ai/docs/operators/llm_file_analysis.rst +++ b/providers/common/ai/docs/operators/llm_file_analysis.rst @@ -159,8 +159,8 @@ Parameters downstream consumer needs the dict shape. This operator also inherits ``LLMOperator``'s HITL review parameters -- -``require_approval``, ``approval_timeout``, ``on_approval_timeout``, and -``allow_modifications`` -- see :doc:`llm` for details. +``require_approval``, ``approval_timeout``, ``on_approval_timeout``, +``allow_modifications``, and ``approval_notifiers`` -- see :doc:`llm` for details. Supported Formats ----------------- diff --git a/providers/common/ai/docs/operators/llm_schema_compare.rst b/providers/common/ai/docs/operators/llm_schema_compare.rst index a8c3c29deec..426fe5ea228 100644 --- a/providers/common/ai/docs/operators/llm_schema_compare.rst +++ b/providers/common/ai/docs/operators/llm_schema_compare.rst @@ -132,8 +132,8 @@ expire with the default ``on_approval_timeout="fail"``, fails the task: returning a ``Sequence[UserContent]`` raises ``TypeError`` before the LLM call. -``approval_timeout``, ``on_approval_timeout``, ``allow_modifications``, and -the rest of the approval behaviour are inherited from +``approval_timeout``, ``on_approval_timeout``, ``allow_modifications``, +``approval_notifiers``, and the rest of the approval behaviour are inherited from :ref:`LLMOperator `. Conditional ETL Based on Schema Compatibility @@ -197,6 +197,8 @@ Parameters ``require_approval=True`` and a positive ``approval_timeout``. - ``allow_modifications``: If ``True``, the reviewer can edit the result JSON before approving. Default ``False``. +- ``approval_notifiers``: Notifier, or list of notifiers, called once the review + is open. Default ``None``. Logging ------- diff --git a/providers/common/ai/docs/operators/llm_sql.rst b/providers/common/ai/docs/operators/llm_sql.rst index bb328f1e67f..c205fa307e2 100644 --- a/providers/common/ai/docs/operators/llm_sql.rst +++ b/providers/common/ai/docs/operators/llm_sql.rst @@ -150,6 +150,9 @@ modified query is re-validated against the same safety rules automatically. :start-after: [START howto_operator_llm_sql_approval] :end-before: [END howto_operator_llm_sql_approval] +``approval_timeout``, ``approval_notifiers``, and the rest of the approval +behaviour are inherited from :ref:`LLMOperator `. + SQL Safety Validation --------------------- diff --git a/providers/common/ai/src/airflow/providers/common/ai/example_dags/example_llm.py b/providers/common/ai/src/airflow/providers/common/ai/example_dags/example_llm.py index ee1a678ce89..b8d2c48b80b 100644 --- a/providers/common/ai/src/airflow/providers/common/ai/example_dags/example_llm.py +++ b/providers/common/ai/src/airflow/providers/common/ai/example_dags/example_llm.py @@ -25,6 +25,7 @@ from pydantic import BaseModel from pydantic_ai.usage import UsageLimits from airflow.providers.common.ai.operators.llm import LLMOperator +from airflow.providers.common.compat.notifier import BaseNotifier from airflow.providers.common.compat.sdk import dag, task @@ -180,6 +181,17 @@ example_llm_operator_templated_usage_limits() # [START howto_operator_llm_approval] +class LogNotifier(BaseNotifier): + template_fields = ("message",) + + def __init__(self, message: str) -> None: + super().__init__() + self.message = message + + def notify(self, context) -> None: + self.log.info(self.message) + + @dag(tags=["example"]) def example_llm_operator_approval(): @@ -192,6 +204,7 @@ def example_llm_operator_approval(): approval_timeout=timedelta(hours=24), on_approval_timeout="approve", allow_modifications=True, + approval_notifiers=LogNotifier(message="{{ task.subject }}\n{{ task.body }}"), ) diff --git a/providers/common/ai/src/airflow/providers/common/ai/mixins/approval.py b/providers/common/ai/src/airflow/providers/common/ai/mixins/approval.py index 3153eb3a999..9a011a0d2e8 100644 --- a/providers/common/ai/src/airflow/providers/common/ai/mixins/approval.py +++ b/providers/common/ai/src/airflow/providers/common/ai/mixins/approval.py @@ -22,6 +22,7 @@ import logging from datetime import timedelta from typing import TYPE_CHECKING, Any, ClassVar, Literal, Protocol +from jinja2 import TemplateError from pydantic import BaseModel, TypeAdapter from airflow.providers.common.compat.version_compat import AIRFLOW_V_3_3_PLUS @@ -34,6 +35,9 @@ if AIRFLOW_V_3_3_PLUS: log = logging.getLogger(__name__) if TYPE_CHECKING: + from collections.abc import Sequence + + from airflow.providers.common.compat.notifier import BaseNotifier from airflow.sdk import Context @@ -43,6 +47,7 @@ class DeferForApprovalProtocol(Protocol): approval_timeout: timedelta | None allow_modifications: bool on_approval_timeout: Literal["fail", "approve", "reject"] + approval_notifiers: Sequence[BaseNotifier] prompt: str task_id: str defer: Any @@ -69,12 +74,23 @@ class LLMApprovalMixin: task resumes as if a reviewer had chosen it. The chosen option is also pre-highlighted for the reviewer in the HITL form. + ``approval_notifiers`` are called once the review is open, so a reviewer + learns about it without watching the Required Actions page, the way + :class:`~airflow.providers.standard.operators.hitl.HITLOperator` does with + ``notifiers``. The review ``subject`` and ``body`` are exposed as + ``{{ task.subject }}`` and ``{{ task.body }}`` in notifier templates. A + notifier whose delivery raises is logged without failing the task, while a + template error fails the task. A retry re-runs the LLM and re-notifies + with the regenerated output, while the open review keeps the original + subject and body. + Operators that use this mixin must set the following attributes: - ``require_approval`` (``bool``) - ``allow_modifications`` (``bool``) - ``approval_timeout`` (``timedelta | None``) - ``on_approval_timeout`` (``Literal["fail", "approve", "reject"]``) + - ``approval_notifiers`` (``Sequence[BaseNotifier]``) - ``prompt`` (``str``) """ @@ -176,6 +192,16 @@ class LLMApprovalMixin: params=hitl_params, ) + self.subject = subject + self.body = body + for notifier in self.approval_notifiers: + try: + notifier({**context}) + except TemplateError: + raise + except Exception: + log.exception("Approval notifier %s failed; the review stays open", notifier) + if AIRFLOW_V_3_3_PLUS: # New core (3.3+): park the task in AWAITING_INPUT -- no trigger, no triggerer. The # task is resumed by the Core API response handler or the scheduler timeout sweep. diff --git a/providers/common/ai/src/airflow/providers/common/ai/operators/llm.py b/providers/common/ai/src/airflow/providers/common/ai/operators/llm.py index 2de3762de6a..358775b221d 100644 --- a/providers/common/ai/src/airflow/providers/common/ai/operators/llm.py +++ b/providers/common/ai/src/airflow/providers/common/ai/operators/llm.py @@ -18,7 +18,7 @@ from __future__ import annotations -from collections.abc import Sequence +from collections.abc import Iterable, Sequence from datetime import timedelta from functools import cached_property from typing import TYPE_CHECKING, Any, ClassVar, Literal @@ -30,6 +30,7 @@ from airflow.providers.common.ai.mixins.approval import LLMApprovalMixin from airflow.providers.common.ai.utils.logging import log_run_summary from airflow.providers.common.ai.utils.output_type import rehydrate_pydantic_output from airflow.providers.common.ai.utils.usage import coerce_usage_limits +from airflow.providers.common.compat.notifier import BaseNotifier from airflow.providers.common.compat.sdk import BaseOperator try: @@ -111,6 +112,11 @@ class LLMOperator(BaseOperator, LLMApprovalMixin): :param allow_modifications: If ``True``, the reviewer can edit the output before approving. The modified value is returned as the task result. Default ``False``. + :param approval_notifiers: Notifiers called once the review is open, so a + reviewer is told about it. Only takes effect with + ``require_approval=True``. A retry re-notifies with the regenerated + output while the open review keeps the original subject and body. + Default ``None``. :param serialize_output: If ``True`` and ``output_type`` is a Pydantic ``BaseModel`` subclass, the model instance is dumped to a ``dict`` via ``model_dump()`` before being pushed to XCom. Default ``False`` -- @@ -144,6 +150,7 @@ class LLMOperator(BaseOperator, LLMApprovalMixin): approval_timeout: timedelta | None = None, on_approval_timeout: Literal["fail", "approve", "reject"] = "fail", allow_modifications: bool = False, + approval_notifiers: BaseNotifier | Iterable[BaseNotifier] | None = None, serialize_output: bool = False, **kwargs: Any, ) -> None: @@ -177,6 +184,19 @@ class LLMOperator(BaseOperator, LLMApprovalMixin): self.approval_timeout = approval_timeout self.on_approval_timeout = on_approval_timeout self.allow_modifications = allow_modifications + if approval_notifiers is None: + approval_notifiers = [] + elif isinstance(approval_notifiers, BaseNotifier): + approval_notifiers = [approval_notifiers] + elif isinstance(approval_notifiers, str) or not isinstance(approval_notifiers, Iterable): + raise TypeError( + "approval_notifiers must be a BaseNotifier or an iterable of BaseNotifier instances, " + f"got {approval_notifiers!r}" + ) + self.approval_notifiers = list(approval_notifiers) + for notifier in self.approval_notifiers: + if not isinstance(notifier, BaseNotifier): + raise TypeError(f"approval_notifiers must contain BaseNotifier instances, got {notifier!r}") @cached_property def llm_hook(self) -> PydanticAIHook: diff --git a/providers/common/ai/src/airflow/providers/common/ai/operators/llm_branch.py b/providers/common/ai/src/airflow/providers/common/ai/operators/llm_branch.py index e646bc0d8c8..ddaa39996dc 100644 --- a/providers/common/ai/src/airflow/providers/common/ai/operators/llm_branch.py +++ b/providers/common/ai/src/airflow/providers/common/ai/operators/llm_branch.py @@ -62,7 +62,7 @@ class LLMBranchOperator(LLMOperator, BranchMixIn): Human-in-the-Loop approval parameters are inherited from :class:`~airflow.providers.common.ai.operators.llm.LLMOperator` (``require_approval``, ``approval_timeout``, ``on_approval_timeout``, - ``allow_modifications``). + ``allow_modifications``, ``approval_notifiers``). The task pauses after the LLM chooses the branch(es) and only skips the unselected downstream tasks once a reviewer approves. Rejecting the review skips the direct downstream tasks except teardowns, matching diff --git a/providers/common/ai/src/airflow/providers/common/ai/operators/llm_file_analysis.py b/providers/common/ai/src/airflow/providers/common/ai/operators/llm_file_analysis.py index c671587a3d0..66d31d2eb59 100644 --- a/providers/common/ai/src/airflow/providers/common/ai/operators/llm_file_analysis.py +++ b/providers/common/ai/src/airflow/providers/common/ai/operators/llm_file_analysis.py @@ -73,7 +73,7 @@ class LLMFileAnalysisOperator(LLMOperator): Human-in-the-Loop approval parameters are inherited from :class:`~airflow.providers.common.ai.operators.llm.LLMOperator` (``require_approval``, ``approval_timeout``, ``on_approval_timeout``, - ``allow_modifications``). + ``allow_modifications``, ``approval_notifiers``). The task pauses after the file analysis and only returns the result once a reviewer approves. """ diff --git a/providers/common/ai/src/airflow/providers/common/ai/operators/llm_schema_compare.py b/providers/common/ai/src/airflow/providers/common/ai/operators/llm_schema_compare.py index 9e5127f8da9..aa25bf7229d 100644 --- a/providers/common/ai/src/airflow/providers/common/ai/operators/llm_schema_compare.py +++ b/providers/common/ai/src/airflow/providers/common/ai/operators/llm_schema_compare.py @@ -112,7 +112,7 @@ class LLMSchemaCompareOperator(LLMOperator): Human-in-the-Loop approval parameters are inherited from :class:`~airflow.providers.common.ai.operators.llm.LLMOperator` (``require_approval``, ``approval_timeout``, ``on_approval_timeout``, - ``allow_modifications``). + ``allow_modifications``, ``approval_notifiers``). The task pauses after the comparison and only returns the result once a reviewer approves. The review body shows the compatibility verdict, a mismatch severity summary, and the full result JSON. diff --git a/providers/common/ai/src/airflow/providers/common/ai/operators/llm_sql.py b/providers/common/ai/src/airflow/providers/common/ai/operators/llm_sql.py index 87a81e0882c..4e22fdc845e 100644 --- a/providers/common/ai/src/airflow/providers/common/ai/operators/llm_sql.py +++ b/providers/common/ai/src/airflow/providers/common/ai/operators/llm_sql.py @@ -85,7 +85,7 @@ class LLMSQLQueryOperator(LLMOperator): Human-in-the-Loop approval parameters are inherited from :class:`~airflow.providers.common.ai.operators.llm.LLMOperator` (``require_approval``, ``approval_timeout``, ``on_approval_timeout``, - ``allow_modifications``). + ``allow_modifications``, ``approval_notifiers``). When ``allow_modifications=True`` and the reviewer edits the SQL, the modified query is re-validated against the same safety rules before being returned. diff --git a/providers/common/ai/tests/unit/common/ai/mixins/test_approval.py b/providers/common/ai/tests/unit/common/ai/mixins/test_approval.py index 6c9f55224a6..c9f18c6b188 100644 --- a/providers/common/ai/tests/unit/common/ai/mixins/test_approval.py +++ b/providers/common/ai/tests/unit/common/ai/mixins/test_approval.py @@ -23,16 +23,20 @@ from tests_common.test_utils.version_compat import AIRFLOW_V_3_1_PLUS, AIRFLOW_V if not AIRFLOW_V_3_1_PLUS: pytest.skip("Human in the loop is only compatible with Airflow >= 3.1.0", allow_module_level=True) +from collections.abc import Sequence from datetime import timedelta from unittest.mock import MagicMock, patch from uuid import uuid4 +from jinja2 import TemplateError from pydantic import BaseModel from airflow.providers.common.ai.mixins.approval import ( LLMApprovalMixin, ) +from airflow.providers.common.compat.notifier import BaseNotifier from airflow.providers.standard.exceptions import HITLRejectException, HITLTriggerEventError +from airflow.sdk import DAG if AIRFLOW_V_3_3_PLUS: from airflow.sdk.exceptions import TaskAwaitingInput @@ -43,6 +47,19 @@ UTCNOW_PATH = "airflow.sdk.timezone.utcnow" AWAIT_INPUT_FLAG_PATH = "airflow.providers.common.ai.mixins.approval.AIRFLOW_V_3_3_PLUS" +class RecordingNotifier(BaseNotifier): + template_fields: Sequence[str] = ("subject", "message") + + def __init__(self, *, subject: str | None = None, message: str = "{{ task.subject }}: {{ task.body }}"): + super().__init__() + self.subject = subject + self.message = message + self.sent: list[str] = [] + + def notify(self, context): + self.sent.append(self.message) + + class FakeOperator(LLMApprovalMixin): """Minimal concrete class satisfying both mixin protocols.""" @@ -54,12 +71,14 @@ class FakeOperator(LLMApprovalMixin): approval_timeout: timedelta | None = None, on_approval_timeout: str = "fail", allow_modifications: bool = False, + approval_notifiers: Sequence[BaseNotifier] = (), ): self.prompt = prompt self.task_id = task_id self.approval_timeout = approval_timeout self.on_approval_timeout = on_approval_timeout self.allow_modifications = allow_modifications + self.approval_notifiers = approval_notifiers self.defer = MagicMock() self.log = MagicMock() @@ -79,7 +98,7 @@ def approval_op_with_modifications(): def context(): ti = MagicMock() ti.id = uuid4() - return MagicMock(**{"__getitem__": lambda self, key: {"task_instance": ti}[key]}) + return {"task_instance": ti, "dag": DAG("test_dag")} # The legacy trigger path is taken on cores < 3.3; pin the flag so these tests keep @@ -176,6 +195,66 @@ class TestDeferForApproval: defer_kwargs = approval_op_with_modifications.defer.call_args[1] assert defer_kwargs["kwargs"]["generated_output"] == '["task_a"]' + @patch(HITL_TRIGGER_PATH, autospec=True) + @patch(UPSERT_HITL_PATH) + def test_notifiers_fire_once_the_review_is_open(self, mock_upsert, mock_trigger_cls, context): + notifier = MagicMock(spec=BaseNotifier) + order = MagicMock() + order.attach_mock(mock_upsert, "open_review") + order.attach_mock(notifier, "notify") + op = FakeOperator(approval_notifiers=[notifier]) + + op.defer_for_approval(context, "output") + + notifier.assert_called_once_with(context) + assert [call[0] for call in order.mock_calls] == ["open_review", "notify"] + + @patch(HITL_TRIGGER_PATH, autospec=True) + @patch(UPSERT_HITL_PATH) + def test_notifier_renders_review_subject_and_body_despite_own_template_fields( + self, mock_upsert, mock_trigger_cls, context + ): + notifier = RecordingNotifier() + op = FakeOperator(approval_notifiers=[notifier]) + context["task"] = op + + op.defer_for_approval(context, "output") + + assert notifier.sent == [ + "Review output for task `test_task`: ```\nPrompt: Summarize this\n\noutput\n```" + ] + assert set(context) == {"task_instance", "dag", "task"} + + @pytest.mark.parametrize("message", ["{{ 'x' | no_such_filter }}", "{{ task.bodyy }}"]) + @patch(HITL_TRIGGER_PATH, autospec=True) + @patch(UPSERT_HITL_PATH) + def test_notifier_template_error_fails_the_task(self, mock_upsert, mock_trigger_cls, context, message): + notifier = RecordingNotifier(message=message) + op = FakeOperator(approval_notifiers=[notifier]) + + with pytest.raises(TemplateError): + op.defer_for_approval(context, "output") + + op.defer.assert_not_called() + + @patch(HITL_TRIGGER_PATH, autospec=True) + @patch(UPSERT_HITL_PATH) + @patch("airflow.providers.common.ai.mixins.approval.log", autospec=True) + def test_notifier_failure_does_not_stop_the_review( + self, mock_log, mock_upsert, mock_trigger_cls, context + ): + failing = MagicMock(spec=BaseNotifier, side_effect=RuntimeError("smtp down")) + healthy = MagicMock(spec=BaseNotifier) + op = FakeOperator(approval_notifiers=[failing, healthy]) + + op.defer_for_approval(context, "output") + + mock_log.exception.assert_called_once_with( + "Approval notifier %s failed; the review stays open", failing + ) + healthy.assert_called_once_with(context) + op.defer.assert_called_once() + @patch(HITL_TRIGGER_PATH, autospec=True) @patch(UPSERT_HITL_PATH) def test_no_modifications_params_empty(self, mock_upsert, mock_trigger_cls, approval_op, context): @@ -512,6 +591,16 @@ class TestAwaitInputForApproval: assert mock_upsert.call_args[1]["options"] == ["Approve", "Reject"] approval_op.defer.assert_not_called() + @patch(UPSERT_HITL_PATH) + def test_notifiers_fire_before_awaiting_input(self, mock_upsert, context): + notifier = MagicMock(spec=BaseNotifier) + op = FakeOperator(approval_notifiers=[notifier]) + + with pytest.raises(TaskAwaitingInput): + op.defer_for_approval(context, "output") + + notifier.assert_called_once() + @patch(UPSERT_HITL_PATH) def test_approval_timeout_carried_on_await(self, mock_upsert, context): timeout = timedelta(hours=2) diff --git a/providers/common/ai/tests/unit/common/ai/operators/test_llm.py b/providers/common/ai/tests/unit/common/ai/operators/test_llm.py index 5ed69546493..640e96640f4 100644 --- a/providers/common/ai/tests/unit/common/ai/operators/test_llm.py +++ b/providers/common/ai/tests/unit/common/ai/operators/test_llm.py @@ -41,6 +41,7 @@ try: except ImportError: _CORE_WALKER = False +from airflow.providers.common.compat.notifier import BaseNotifier from airflow.providers.common.compat.sdk import TaskDeferred if AIRFLOW_V_3_3_PLUS: @@ -327,6 +328,7 @@ class TestLLMOperatorApproval: assert op.allow_modifications is False assert op.approval_timeout is None assert op.on_approval_timeout == "fail" + assert op.approval_notifiers == [] def test_unknown_on_approval_timeout_raises(self): with pytest.raises(ValueError, match="on_approval_timeout must be"): @@ -359,6 +361,31 @@ class TestLLMOperatorApproval: ): LLMOperator(task_id="t", prompt="p", llm_conn_id="c", on_approval_timeout="approve", **kwargs) + def test_single_approval_notifier_normalized_to_list(self): + notifier = MagicMock(spec=BaseNotifier) + op = LLMOperator(task_id="t", prompt="p", llm_conn_id="c", approval_notifiers=notifier) + assert op.approval_notifiers == [notifier] + + @pytest.mark.parametrize( + ("notifiers", "match"), + [ + (print, r"iterable of BaseNotifier instances, got "), + (5, r"iterable of BaseNotifier instances, got 5"), + ("not-a-notifier", r"iterable of BaseNotifier instances, got 'not-a-notifier'"), + ({"a": 1}, r"must contain BaseNotifier instances, got 'a'"), + ([object()], r"must contain BaseNotifier instances, got