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

@@ -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
 -------

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 <howto/operator:llm>`.
 
 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``.

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
 -----------------

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 <howto/operator:llm>`.
 
 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
 -------

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 <howto/operator:llm>`.
+
 SQL Safety Validation
 ---------------------
 

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 }}"),
     )
 
 

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.

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:

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

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.
     """

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.

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.

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)

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 <built-in function print>"),
+            (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 <object object"),
+        ],
+        ids=["callable", "int", "str", "dict", "list_of_object"],
+    )
+    def test_rejects_non_notifier_approval_notifiers(self, notifiers, match):
+        with pytest.raises(TypeError, match=match):
+            LLMOperator(task_id="t", prompt="p", llm_conn_id="c", approval_notifiers=notifiers)
+
+    def test_accepts_generator_of_approval_notifiers(self):
+        notifiers = [MagicMock(spec=BaseNotifier), MagicMock(spec=BaseNotifier)]
+        op = LLMOperator(task_id="t", prompt="p", llm_conn_id="c", approval_notifiers=iter(notifiers))
+        assert op.approval_notifiers == notifiers
+
     @patch("airflow.providers.standard.triggers.hitl.HITLTrigger", autospec=True)
     @patch("airflow.sdk.execution_time.hitl.upsert_hitl_detail")
     @patch("airflow.providers.common.ai.operators.llm.PydanticAIHook", autospec=True)