kaxil commented on code in PR #72159:
URL: https://github.com/apache/airflow/pull/72159#discussion_r4004741545
##########
providers/common/ai/src/airflow/providers/common/ai/mixins/approval.py:
##########
@@ -165,6 +177,14 @@ def defer_for_approval(
params=hitl_params,
)
+ self.subject = subject
+ self.body = body
+ for notifier in self.approval_notifiers:
+ try:
+ notifier(context)
Review Comment:
Two things about this loop, the first with a real consequence.
`BaseNotifier.__call__` runs `_update_context` and `render_template_fields`
outside its own try (`notifier.py:135-141`), so a template error surfaces here
rather than inside the notifier, and `DAG.template_undefined` defaults to
`jinja2.StrictUndefined` (`dag.py:472`). Measured on a DAG-bound operator: `{{
task.subject }}` sends, while `{{ task.bodyy }}` and `{{ nonexistent_var }}`
each raise `UndefinedError`, get swallowed here, and leave `sent == []` with
the review opened and the task parked. So a one-character typo sends nothing,
deterministically, on every run. `HITLDetail` has no delivery-status column
(`models/hitl.py:140-164`), so the Required Actions page looks like any healthy
pending review, the only trace is two ERROR lines in the task log, and with
`approval_timeout=None` (the documented default) it waits indefinitely -- which
is the pre-PR situation from this PR's own Why, reached through the commonest au
thoring mistake. I asked for the swallow in round 1 and am not asking for it
back: a transient SMTP outage should not fail the task. The point is that one
`try` spans both framework rendering and user-code delivery while only delivery
is plausibly transient -- should the swallow be scoped to `notify`, or the
permanence at least be made visible in the log? Separately and much smaller:
206112f8 passed `{**context, ...}` here and this commit swapped in the live
dict, so each notifier's `template_fields` now land in the dict the next one
renders against. Reproduced, though it needs a later notifier to reference a
bare name an earlier one declares, which no stock notifier does, and
`HITLOperator` hands over the live context too -- the only reason to mention it
is that keeping the copy cost nothing.
##########
providers/common/ai/docs/operators/llm.rst:
##########
@@ -205,6 +205,14 @@ approving with ``allow_modifications=True``, and set a
deadline with
: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.
Review Comment:
`{{ task.subject }}` rather than the bare `{{ subject }}` is the
counterintuitive half of this contract, and prose here is the only place it is
stated. No example DAG uses `approval_notifiers` at all, while the operator
this mirrors has one: `example_hitl_operator.py` wires a `LocalLogNotifier`
templating `{{ task.subject }}` / `{{ task.body }}` into every `HITLOperator`.
A couple of lines inside the existing `[START howto_operator_llm_approval]`
block in `example_llm.py` would pin the spelling where people actually copy
from. `llm_sql.rst` has the same shape, and this diff moved it there: the colon
that introduces the `exampleinclude` used to close "re-validated against the
same safety rules automatically:", which the example does show, and now closes
the sentence that was edited to add `approval_notifiers`
(`llm_sql.rst:143-146`) while `[START howto_operator_llm_sql_approval]` sets
`require_approval`, `approval_timeout` and `allow_modifications` and no
notifier. So both pages no
w promise an example of this parameter and render one without it. Users copy
example DAGs verbatim, which is the standing reason to want at least one that
demonstrates the parameter rather than only prose describing it. The gap is
wider than the operator pages too: the five `@task.llm*` decorators each
forward `**kwargs` to their operator, so they accept and honour
`approval_notifiers`, and no decorator doc mentions it. One more line worth
having on this page: the default `body` is the prompt and the output in a code
fence (`approval.py:150`), and `prompt` is a rendered template field, so
anything templated into it is already substituted by the time `{{ task.body }}`
reaches a notifier. The same text already goes to the HITL row, so this is not
new exposure, but the review UI sits inside Airflow's auth boundary and a Slack
channel does not -- worth suggesting `{{ task.subject }}` plus a link when the
prompt or output is sensitive.
##########
providers/common/ai/src/airflow/providers/common/ai/operators/llm.py:
##########
@@ -29,7 +29,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.compat.sdk import BaseOperator
+from airflow.providers.common.compat.sdk import BaseNotifier, BaseOperator
Review Comment:
This makes the module unimportable on Airflow 3.0.0 and 3.0.1. Pre-PR this
line imported only `BaseOperator`. `BaseNotifier` resolves through
`compat/sdk.py:207` as `("airflow.sdk", "airflow.notifications.basenotifier")`,
and I checked both candidates at the release tags: on 3.0.0 and 3.0.1
`airflow.sdk`'s lazy map points `BaseNotifier` at `.definitions.notifier`,
which exists in neither (it was repointed to `.bases.notifier` in 3.0.2), and
`airflow/notifications/basenotifier.py` is absent from every 3.x tag, so the
second candidate is dead across all of 3.x. `_compat_utils` catches both and
then raises `ImportError("Could not import 'BaseNotifier' from any of: ...")`.
The provider declares `apache-airflow>=3.0.0` (`pyproject.toml:70`) with no
import-time gate, and neither `standard>=1.12.1` nor `common-compat>=1.15.0`
lifts that floor (both are `apache-airflow>=2.11.0`), so the combination
installs. This module is the base of all four operator subclasses and the five
`@task.llm*`
decorators, so on those two cores any DAG touching the provider's LLM surface
fails to parse, approvals or not. `from
airflow.providers.common.compat.notifier import BaseNotifier` avoids it -- that
module imports `airflow.sdk.bases.notifier` directly, which is present at
3.0.0. Raising the declared floor past 3.0.1 would be an equally honest answer
if those cores are not really supported; the concern is that the PR currently
decides it silently. (`approval.py:39` has the same import under
`TYPE_CHECKING`, so it is harmless today -- worth moving only for consistency.)
##########
providers/common/ai/src/airflow/providers/common/ai/operators/llm.py:
##########
@@ -138,6 +142,14 @@ def __init__(
self.require_approval = require_approval
self.approval_timeout = approval_timeout
self.allow_modifications = allow_modifications
+ self.approval_notifiers = (
+ [approval_notifiers]
+ if isinstance(approval_notifiers, BaseNotifier)
+ else list(approval_notifiers or [])
Review Comment:
Follow-up on my round-2 ask: the `list(...)` landed, but the guard behind it
only catches a wrong *element*, because normalization runs first. Ran the cases
against this constructor: `approval_notifiers=send_slack_alert` (a bare
callable, and the likeliest mistake given Airflow's callback params take plain
callables) dies on this line as `TypeError: 'function' object is not iterable`
and never reaches the guard; `5` gives `'int' object is not iterable`; a str is
split into characters, so `"not-a-notifier"` reports `got 'n'` and
`"my_notifier"` reports `got 'm'`; a dict iterates keys and reports `got 'a'`.
Only `[object()]` produces the intended message. The test pins the weak half
rather than catching it -- `test_rejects_non_notifier_approval_notifiers`
passes `"not-a-notifier"` and its `match=` stops before the offending value, so
`got 'n'` is green, and no case covers a non-iterable at all. Rejecting the
scalar case before iterating closes both symptoms, with `str` ruled out exp
licitly ahead of that check since `str` is itself `Iterable` and would
otherwise still split. Worth reaching for `Iterable` rather than `Sequence`
there: a `set` and a generator are both accepted today, and a `Sequence` bound
would quietly stop taking them.
##########
providers/common/ai/src/airflow/providers/common/ai/mixins/approval.py:
##########
@@ -165,6 +177,14 @@ def defer_for_approval(
params=hitl_params,
)
+ self.subject = subject
Review Comment:
A reviewer gets notified about one generation and approves another, and the
feature's success condition is the bug's trigger. The docstring above says
notifiers exist so "a reviewer learns about it without watching the Required
Actions page", and the PR's Why is that reviews sat until someone happened to
look, so acting on the notification instead of opening the UI is the intended
workflow. A retry or clear is exactly when the two disagree:
`prepare_db_for_next_try` rotates `task_instance.id` (`taskinstance.py:1076`),
`hitl_detail_ti_fkey` is `onupdate="CASCADE"` (`hitl.py:171-176`) so the review
row follows the new id, and neither upsert branch rewrites
`subject`/`body`/`params` -- the no-response branch updates nothing at all and
the with-response branch clears only the four response columns
(`execution_api/routes/hitl.py:71-93`). Attempt 2's notification therefore
describes its own generation while the review page still shows attempt 1's.
Only the retry has to happen for this t
o bite; the outputs differing is the default for a non-deterministic model,
not a second condition. Aggravated, though pre-existing and not this PR's to
fix: `params` survives too, so with `allow_modifications=True` the form is
pre-filled with attempt 1's output, and approving unedited makes `modified !=
generated_output` true, so `execute_complete` returns attempt 1's text as the
task result (approval.py:282-284). Should the notification carry the subject
and body the review actually holds rather than the freshly computed ones? The
authoritative row is on the wire as `HITLDetailRequestResult`
(`comms.py:788-794`), but `upsert_hitl_detail` discards it and is typed `->
None` (`task-sdk/.../execution_time/hitl.py:39-47`), so consuming it is a Task
SDK change -- if that is out of scope here, a docs note saying a retry
re-notifies with regenerated content while the review keeps the original is
provider-only and gets the same honesty. The retry really does re-run
`execute`: `HITLTimeoutE
rror` extends `HITLTriggerEventError(Exception)`, so an approval timeout is an
ordinary retryable failure, and `clear_next_method_args` unsets
`next_method`/`next_kwargs` so the retry starts from the top
(`taskinstance.py:1572-1577`). A 3.3+ resume and a legacy triggerer restart
both come back through `next_method` instead, so neither re-notifies.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
