kaxil commented on code in PR #72155:
URL: https://github.com/apache/airflow/pull/72155#discussion_r3968181982
##########
providers/common/ai/src/airflow/providers/common/ai/mixins/approval.py:
##########
@@ -179,14 +189,13 @@ def defer_for_approval(
trigger=HITLTrigger(
ti_id=ti_id,
options=[LLMApprovalMixin.APPROVE, LLMApprovalMixin.REJECT],
- defaults=None,
+ defaults=timeout_defaults,
Review Comment:
Separate from the `timeout=` kwarg thread: making `defaults` non-null also
arms a `HITLTrigger` replay path that can turn the default into a hard failure
on cores below 3.3. `_handle_timeout` Case 3 PATCHes the response, and the
execution API sets `responded_by = None` with `responded_at = utcnow()`
(`execution_api/routes/hitl.py:129-130`).
If that trigger is replayed, which a triggerer restart or an
`assign_unassigned` reassignment after a heartbeat lapse will do, `run()`
re-enters `_handle_timeout` and now matches Case 1, since `response_received`
is just `responded_at is not None` (`models/hitl.py:91`) and Case 3 wrote
`chosen_options=self.defaults`. Case 1 then dereferences
`resp.responded_by_user.name` (`standard/triggers/hitl.py:130`) where the value
is `None`; the `assert` above it sits inside `if TYPE_CHECKING`, so nothing
guards it at runtime. The `AttributeError` reaches `Trigger.submit_failure`,
`next_method` becomes `__fail__`, and `execute_complete` never runs.
At the merge base `defaults` was always `None` here, so this was unreachable
before. Guarding those two log arguments in `HITLTrigger` fixes it for
`HITLOperator` too, which has the same exposure whenever `defaults=` is set.
##########
providers/common/ai/docs/operators/llm.rst:
##########
@@ -198,7 +198,11 @@ Set ``require_approval=True`` to pause the task after the
LLM generates its
output and wait for a human reviewer to approve or reject it via the Airflow
HITL interface. Optionally allow the reviewer to edit the output before
approving with ``allow_modifications=True``, and set a deadline with
-``approval_timeout``:
+``approval_timeout``.
+
+When ``approval_timeout`` expires without a review, the task fails by default.
+Set ``on_approval_timeout="approve"`` or ``"reject"`` to answer the review with
+that option instead, so an unattended pipeline keeps moving:
Review Comment:
`"reject"` does not keep the pipeline moving on this operator.
`HITLRejectException` is a plain `AirflowException` and only
`LLMBranchOperator` catches it (`llm_branch.py:149`), so on `LLMOperator`,
`LLMSQLQueryOperator`, `LLMFileAnalysisOperator` and
`LLMSchemaCompareOperator`, `on_approval_timeout="reject"` reaches the same
terminal state as the `"fail"` default: task failed. Only `"approve"` changes
the outcome here.
The `exampleinclude` right below makes it concrete: `example_llm.py:162`
sets `on_approval_timeout="reject"` on a plain `LLMOperator` inside
`howto_operator_llm_approval`, so the page demonstrates the one value that
changes nothing. `llm_branch.rst:100-104` already words this correctly. Worth
splitting the sentence so `"approve"` is described as returning the output and
letting downstream run while `"reject"` still fails the task here, pointing at
`LLMBranchOperator` for the skip behaviour, and switching the example to
`"approve"`.
Nearby: `llm_sql.py`'s docstring lists `on_approval_timeout` but
`llm_sql.rst`'s approval section does not, unlike
`llm_file_analysis.rst:161-163`.
##########
providers/common/ai/src/airflow/providers/common/ai/mixins/approval.py:
##########
@@ -219,10 +229,12 @@ def execute_complete(self, context: Context,
generated_output: str, event: dict[
responded_by_user = event.get("responded_by_user")
chosen = event["chosen_options"]
if self.APPROVE not in chosen:
+ if event.get("timedout"):
+ raise HITLRejectException("Output was rejected by the approval
timeout default.")
Review Comment:
This message names neither the setting nor the deadline. Something like
`"Output was rejected automatically: approval_timeout expired with
on_approval_timeout='reject'."` would tell the operator what happened.
The approve side has the mirror gap: `execute_complete` returns silently, so
a task that succeeded because nobody reviewed it looks exactly like one a human
approved (`ApprovalOperator` logs `Approved. Proceeding with downstream
tasks...` for this reason, `standard/operators/hitl.py:459`). A `log.info`
naming `responded_by_user or "the approval timeout default"` would close it,
via the module-level `log` at line 34 since the mixin is not a `BaseOperator`.
Related: `llm_branch.py:152` now renders `Rejected by None. Skipping
downstream tasks...`, since both wait paths leave `responded_by_user=None` on a
timeout default.
##########
providers/common/ai/src/airflow/providers/common/ai/mixins/approval.py:
##########
@@ -42,6 +42,7 @@ class DeferForApprovalProtocol(Protocol):
approval_timeout: timedelta | None
allow_modifications: bool
+ on_approval_timeout: str
Review Comment:
The Protocol types this `str` while `LLMOperator.__init__` types it
`Literal["fail", "approve", "reject"]` (`llm.py:126`). Every other member
mirrors its concrete type (`defer: Any` aside), and the value is only used as a
dict key, so the same `Literal` here would let mypy catch a bad value at the
mixin boundary.
--
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]
