This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new d318d5484f3 Fail LLMOperator approval at parse time on Airflow cores
older than 3.1 (#73261)
d318d5484f3 is described below
commit d318d5484f3db618b84534854ce658ab034405b2
Author: Kaxil Naik <[email protected]>
AuthorDate: Thu Sep 17 14:04:26 2026 +0100
Fail LLMOperator approval at parse time on Airflow cores older than 3.1
(#73261)
Human-in-the-loop review needs Airflow 3.1+, but require_approval=True had
no
version guard. On a 3.0.x core, which apache-airflow>=3.0.0 supports, the
Dag
imported fine and the task ran the model call before defer_for_approval
failed
on an import that cannot succeed there. Every retry paid for the call again.
Reject the parameter in __init__ instead, matching how AgentOperator already
gates enable_hitl_review.
---
providers/common/ai/docs/operators/llm.rst | 8 +++-
.../airflow/providers/common/ai/operators/llm.py | 10 ++++-
.../ai/tests/unit/common/ai/decorators/test_llm.py | 3 ++
.../unit/common/ai/decorators/test_llm_branch.py | 3 ++
.../ai/decorators/test_llm_schema_compare.py | 3 ++
.../unit/common/ai/decorators/test_llm_sql.py | 3 ++
.../ai/tests/unit/common/ai/operators/test_llm.py | 52 +++++++++++++++++++++-
7 files changed, 78 insertions(+), 4 deletions(-)
diff --git a/providers/common/ai/docs/operators/llm.rst
b/providers/common/ai/docs/operators/llm.rst
index ecbbfe3584a..ce0b2e18784 100644
--- a/providers/common/ai/docs/operators/llm.rst
+++ b/providers/common/ai/docs/operators/llm.rst
@@ -240,6 +240,12 @@ HITL interface. Optionally allow the reviewer to edit the
output before
approving with ``allow_modifications=True``, and set a deadline with
``approval_timeout``.
+Human-in-the-loop review needs Airflow 3.1+. On an older core the operator
raises
+``AirflowOptionalProviderFeatureException`` when it is constructed, so the Dag
file
+fails to import, and with it every Dag defined in that file. A dynamically
mapped
+task (``.expand()``) is only constructed when it runs, so there the same error
+surfaces as a task failure -- still before the model is called.
+
When ``approval_timeout`` expires without a review, the task fails by default.
Set ``on_approval_timeout="approve"`` to return the generated output instead,
so
an unattended pipeline keeps moving. ``"reject"`` answers the review with a
@@ -287,7 +293,7 @@ Parameters
Fails the task when token / request / tool-call budgets are exceeded, or
when a
templated dict value cannot be coerced. Default ``None``.
- ``require_approval``: If ``True``, the task defers after generating output
and waits
- for human review. Default ``False``.
+ for human review. Default ``False``. Needs Airflow 3.1+.
- ``approval_timeout``: Maximum time to wait for a review (``timedelta``).
``None``
means wait indefinitely. Default ``None``.
- ``on_approval_timeout``: Outcome when ``approval_timeout`` expires without a
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 358775b221d..fb813b7744f 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
@@ -31,7 +31,8 @@ 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
+from airflow.providers.common.compat.sdk import
AirflowOptionalProviderFeatureException, BaseOperator
+from airflow.providers.common.compat.version_compat import AIRFLOW_V_3_1_PLUS
try:
# New enough cores register an operator's declared ``output_type`` classes
for
@@ -99,7 +100,7 @@ class LLMOperator(BaseOperator, LLMApprovalMixin):
caveats.
:param require_approval: If ``True``, the task defers after generating
output and waits for a human reviewer to approve or reject via the
- HITL interface. Default ``False``.
+ HITL interface. Default ``False``. Needs Airflow 3.1+.
:param approval_timeout: Maximum time to wait for a review. When
exceeded, ``on_approval_timeout`` decides the outcome.
:param on_approval_timeout: What to do when ``approval_timeout`` expires
@@ -172,6 +173,11 @@ class LLMOperator(BaseOperator, LLMApprovalMixin):
raise ValueError(
f"on_approval_timeout must be 'fail', 'approve', or 'reject',
got {on_approval_timeout!r}."
)
+ # Checked before the combination rule so an old core reports the core
version
+ # rather than sending the user to drop an argument that was never the
problem.
+ if require_approval and not AIRFLOW_V_3_1_PLUS:
+ raise
AirflowOptionalProviderFeatureException("require_approval=True needs Airflow
3.1+.")
+
if on_approval_timeout != "fail" and not (
require_approval and approval_timeout is not None and
approval_timeout > timedelta(0)
):
diff --git a/providers/common/ai/tests/unit/common/ai/decorators/test_llm.py
b/providers/common/ai/tests/unit/common/ai/decorators/test_llm.py
index 480738617dd..e68d6ce6a78 100644
--- a/providers/common/ai/tests/unit/common/ai/decorators/test_llm.py
+++ b/providers/common/ai/tests/unit/common/ai/decorators/test_llm.py
@@ -23,6 +23,8 @@ from pydantic_ai.messages import ImageUrl
from airflow.providers.common.ai.decorators.llm import _LLMDecoratedOperator
+from tests_common.test_utils.version_compat import AIRFLOW_V_3_1_PLUS
+
class TestLLMDecoratedOperator:
def test_custom_operator_name(self):
@@ -79,6 +81,7 @@ class TestLLMDecoratedOperator:
assert op.prompt == prompt
mock_agent.run_sync.assert_called_once_with(prompt, usage_limits=None)
+ @pytest.mark.skipif(not AIRFLOW_V_3_1_PLUS, reason="require_approval needs
Airflow >= 3.1.0")
@patch("airflow.providers.common.ai.operators.llm.PydanticAIHook",
autospec=True)
def
test_sequence_prompt_with_require_approval_raises_before_run_sync(self,
mock_hook_cls):
"""Sequence prompt + require_approval=True fails before the agent
runs."""
diff --git
a/providers/common/ai/tests/unit/common/ai/decorators/test_llm_branch.py
b/providers/common/ai/tests/unit/common/ai/decorators/test_llm_branch.py
index c832661bee4..3ff69936261 100644
--- a/providers/common/ai/tests/unit/common/ai/decorators/test_llm_branch.py
+++ b/providers/common/ai/tests/unit/common/ai/decorators/test_llm_branch.py
@@ -25,6 +25,8 @@ from pydantic_ai.messages import ImageUrl
from airflow.providers.common.ai.decorators.llm_branch import
_LLMBranchDecoratedOperator
from airflow.providers.common.ai.operators.llm_branch import LLMBranchOperator
+from tests_common.test_utils.version_compat import AIRFLOW_V_3_1_PLUS
+
class TestLLMBranchDecoratedOperator:
def test_custom_operator_name(self):
@@ -73,6 +75,7 @@ class TestLLMBranchDecoratedOperator:
with pytest.raises(TypeError, match="must be"):
op.execute(context={})
+ @pytest.mark.skipif(not AIRFLOW_V_3_1_PLUS, reason="require_approval needs
Airflow >= 3.1.0")
@patch("airflow.providers.common.ai.operators.llm.PydanticAIHook",
autospec=True)
def
test_sequence_prompt_with_require_approval_raises_before_run_sync(self,
mock_hook_cls):
"""Sequence prompt + require_approval=True fails before the agent
runs."""
diff --git
a/providers/common/ai/tests/unit/common/ai/decorators/test_llm_schema_compare.py
b/providers/common/ai/tests/unit/common/ai/decorators/test_llm_schema_compare.py
index 87ca8a10a0d..b56e4eed118 100644
---
a/providers/common/ai/tests/unit/common/ai/decorators/test_llm_schema_compare.py
+++
b/providers/common/ai/tests/unit/common/ai/decorators/test_llm_schema_compare.py
@@ -27,6 +27,8 @@ from airflow.providers.common.ai.operators.llm_schema_compare
import (
SchemaCompareResult,
)
+from tests_common.test_utils.version_compat import AIRFLOW_V_3_1_PLUS
+
def _make_compare_result():
return SchemaCompareResult(
@@ -114,6 +116,7 @@ class TestLLMSchemaCompareDecoratedOperator:
forwarded_prompt = mock_agent.run_sync.call_args[0][0]
assert forwarded_prompt == prompt
+ @pytest.mark.skipif(not AIRFLOW_V_3_1_PLUS, reason="require_approval needs
Airflow >= 3.1.0")
@patch("airflow.providers.common.ai.operators.llm.PydanticAIHook",
autospec=True)
@patch.object(LLMSchemaCompareOperator, "_build_schema_context",
return_value="mocked schema")
def test_sequence_prompt_with_require_approval_raises_before_run_sync(
diff --git
a/providers/common/ai/tests/unit/common/ai/decorators/test_llm_sql.py
b/providers/common/ai/tests/unit/common/ai/decorators/test_llm_sql.py
index 16b85f094a7..099b4118c78 100644
--- a/providers/common/ai/tests/unit/common/ai/decorators/test_llm_sql.py
+++ b/providers/common/ai/tests/unit/common/ai/decorators/test_llm_sql.py
@@ -23,6 +23,8 @@ from pydantic_ai.messages import ImageUrl
from airflow.providers.common.ai.decorators.llm_sql import
_LLMSQLDecoratedOperator
+from tests_common.test_utils.version_compat import AIRFLOW_V_3_1_PLUS
+
class TestLLMSQLDecoratedOperator:
def test_custom_operator_name(self):
@@ -79,6 +81,7 @@ class TestLLMSQLDecoratedOperator:
assert op.prompt == prompt
mock_agent.run_sync.assert_called_once_with(prompt, usage_limits=None)
+ @pytest.mark.skipif(not AIRFLOW_V_3_1_PLUS, reason="require_approval needs
Airflow >= 3.1.0")
@patch("airflow.providers.common.ai.operators.llm.PydanticAIHook",
autospec=True)
def
test_sequence_prompt_with_require_approval_raises_before_run_sync(self,
mock_hook_cls):
"""Sequence prompt + require_approval=True fails before the agent
runs."""
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 640e96640f4..3bed2f0f16f 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
@@ -42,7 +42,7 @@ except ImportError:
_CORE_WALKER = False
from airflow.providers.common.compat.notifier import BaseNotifier
-from airflow.providers.common.compat.sdk import TaskDeferred
+from airflow.providers.common.compat.sdk import
AirflowOptionalProviderFeatureException, TaskDeferred
if AIRFLOW_V_3_3_PLUS:
# On 3.3+ cores require_approval pauses the task in AWAITING_INPUT; older
cores defer
@@ -313,6 +313,56 @@ def _make_context(ti_id=None):
return MagicMock(**{"__getitem__": lambda self, key: {"task_instance":
ti}[key]})
+class TestLLMOperatorApprovalVersionGate:
+ """__init__ rejects require_approval on cores without human-in-the-loop
support.
+
+ Deliberately carries no class-level 3.1 skipif. These tests simulate an
old core by
+ patching the flag, so they must not inherit the sibling class's skip --
and on a
+ genuine pre-3.1 core, such as the 3.0.6 providers-compatibility job, they
are the
+ only tests that exercise the gate natively.
+ """
+
+ @pytest.mark.parametrize(
+ ("kwargs", "expected_exception", "match"),
+ [
+ pytest.param(
+ {"require_approval": True},
+ AirflowOptionalProviderFeatureException,
+ "Airflow 3.1",
+ id="require-approval-rejected",
+ ),
+ pytest.param(
+ {"require_approval": True, "on_approval_timeout": "approve"},
+ AirflowOptionalProviderFeatureException,
+ "Airflow 3.1",
+ id="version-beats-combination-rule",
+ ),
+ pytest.param(
+ {"require_approval": True, "on_approval_timeout": "nope"},
+ ValueError,
+ "on_approval_timeout must be",
+ id="literal-check-keeps-precedence",
+ ),
+ ],
+ )
+ @patch("airflow.providers.common.ai.operators.llm.AIRFLOW_V_3_1_PLUS",
False)
+ def test_old_core_reports_the_blocking_argument(self, kwargs,
expected_exception, match):
+ """Which of two applicable errors __init__ reports, and in which order.
+
+ Dropping on_approval_timeout would not make the operator work on an
older core,
+ so the version has to beat the combination rule. A bad literal is
wrong on every
+ core, so it keeps its own precise message -- which also pins the guard
below the
+ literal check, since hoisting it would swap that message for the
version one.
+ """
+ with pytest.raises(expected_exception, match=match):
+ LLMOperator(task_id="t", prompt="p", llm_conn_id="c", **kwargs)
+
+ @patch("airflow.providers.common.ai.operators.llm.AIRFLOW_V_3_1_PLUS",
False)
+ def test_operator_without_approval_builds_on_old_core(self):
+ op = LLMOperator(task_id="t", prompt="p", llm_conn_id="c")
+ assert op.require_approval is False
+
+
@pytest.mark.skipif(
not AIRFLOW_V_3_1_PLUS, reason="Human in the loop is only compatible with
Airflow >= 3.1.0"
)
