Update WorkflowTrigger to forward failed_stat

[guan404ming]

Related Issue

closes #41140
cc @eladkal

Why

• prevents infinite polling when monitored failure
• correctly propagates failure status back to calling workflow

How

• fix WorkflowTrigger to continue checking allowed_states when no failed states are found
• update ExternalTaskSensor.execute_complete to properly handle "failed" status
• add tests to verify behaviors

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[eladkal]

Nice

[Lee-W]

Looks like we might need to fix some compat issue 🤔

[guan404ming]

Pushed for adding test for exception.

[guan404ming]

Mark it draft for fixing failing ci and would reopen when it is ready.

[guan404ming]

CI failure seems fixed in #50521, reopen it for review.

[Lee-W]

It looks fantastic! Thank you for clarifying the exceptions so much.

[eladkal]

static checks fails

[guan404ming]

I've fixed it.

[guan404ming]

Thanks for all reviews and suggestions!

[gopidesupavan]

I am confused this why this else removed?
What if there is no allowed_states provided and only provided failed_states? i believe it goes infinite loop.

[guan404ming]

Thanks for pointing out.

The else statement was removed to ensure that even if no tasks are found in a failed state (failed_count == 0), the trigger proceeds to check the allowed_states. Previously, it would immediately yield a "success" event and exit, bypassing the allowed_states check.

But in the case you've mentioned it would get into infinite loop, originally I've considered that case too. But I think "success" in this context to "the specified failure condition was not met." not really "success". But it do cause issue for users to get into the infinite loop

To address the continuous polling when only failed_states are specified and no failure occurs, maybe we could add this check below to prevent get into the loop. wdyt?

elif not self.allowed_states and not self.skipped_states:
      yield TriggerEvent({"status": "success"}) # Or a new status like "no_failure_detected"
      return