kaxil commented on code in PR #71073:
URL: https://github.com/apache/airflow/pull/71073#discussion_r3740565295
##########
providers/common/ai/src/airflow/providers/common/ai/operators/llm_branch.py:
##########
@@ -133,7 +143,13 @@ def execute(self, context: Context) -> str | Iterable[str]
| None:
def execute_complete(self, context: Context, generated_output: str, event:
dict[str, Any]) -> Any:
"""Resume after human review, validating the reviewed choice before
branching."""
- output = super().execute_complete(context, generated_output, event)
+ try:
+ output = super().execute_complete(context, generated_output, event)
+ except HITLRejectException:
+ if self.fail_on_reject:
+ raise
+ self.log.info("Rejected. Skipping all downstream tasks...")
Review Comment:
The exception being swallowed here carries the reviewer name (`Output was
rejected by the reviewer <user>.`), and the task log was the only place that
surfaced. `event["responded_by_user"]` is right here, so
`self.log.info("Rejected by %s. Skipping all downstream tasks.",
event.get("responded_by_user"))` would keep the attribution. Right now the log
for a rejected gate no longer says who rejected it.
##########
providers/common/ai/src/airflow/providers/common/ai/operators/llm_branch.py:
##########
@@ -133,7 +143,13 @@ def execute(self, context: Context) -> str | Iterable[str]
| None:
def execute_complete(self, context: Context, generated_output: str, event:
dict[str, Any]) -> Any:
"""Resume after human review, validating the reviewed choice before
branching."""
- output = super().execute_complete(context, generated_output, event)
+ try:
+ output = super().execute_complete(context, generated_output, event)
+ except HITLRejectException:
+ if self.fail_on_reject:
+ raise
+ self.log.info("Rejected. Skipping all downstream tasks...")
+ return self.do_branch(context, None)
Review Comment:
`do_branch(context, None)` skips every direct downstream task, teardowns
included, but `ApprovalOperator`'s reject path filters them out (`yield from (t
for t in tasks if not t.is_teardown)`). I checked this on a real DAG with
`branch >> [task_a, task_b, cleanup.as_teardown()]`: this path skips
`['cleanup', 'task_a', 'task_b']` where `ApprovalOperator` skips `['task_a',
'task_b']`. Before this change a reject failed the task and the teardown still
ran on its `all_done`-style rule, so cleanup now silently stops running after a
rejection.
Either filter teardowns here, or drop the "matching `ApprovalOperator`"
wording from the docstring and the rst.
--
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]
