Support running State.UPSTREAM_FAILED tasks when WATCHER consumer upstream tasks fail

[tatiana]

This PR extends the implementation of ExecutionMode.WATCHER failure handling in #2040 to tackle running State.UPSTREAM_FAILED tasks.

And aims to offer a solution for the scenario described in:
#2040 (comment)

@pankajastro In the case of Jaffle Shop, let's say the following happens:

The dbt_producer_watcher fails while trying to apply the seed raw_customers.
The task raw_customers, which was actively watching, will fail (it becomes red). It will have logs for a first try that was unsuccessful. If it is retried, that would be its second try.
However, the task stg_customers will have never been started; it would never have been initialised as a sensor. It will likely be marked as upstream_failed (I'm asking you to confirm this). If this hypothesis is correct, it will likely never be attempted, and therefore, no logs will be generated for this nonexistent first attempt.

So, my concern is—and this is what I'm asking you to test and confirm—what happens when someone clears the status of the seed raw_customers and its downstream tasks? Let's say it succeeds; the task stg_customers will then proceed to its second attempt, as expected, and execute dbt run. However, I'm worried that the try number on the customers task would be 1 (afterall, it is the first time it is actually being executed), and it would enter sensor mode instead of running the dbt run, and it would wait "forever" (until timeout happens) for an update from the dbt producer that will never come.

Previously - in #2040 - if the status of the seed raw_customers was cleared and that task was run sucessfully falling back to ExecutionMode.LOCAL, the downstream tasks would fail. This happened because it was their first try, and they'd see the producer task had failed, and they would fail, giving a poor user-experience.

With the current PR, we attempt to understand if a task is a State.UPSTREAM_FAILED task based on the newly introduced property poke_retry_number.

Relates to: #1960

[pankajkoti]

The conditionals and approach look good to me now. Happy to merge once CI is green.

[Copilot]

Pull Request Overview

This PR extends failure handling in the WATCHER execution mode to support running tasks that previously failed with State.UPSTREAM_FAILED. When upstream producer tasks fail, consumer tasks can now attempt to run locally using ExecutionMode.LOCAL instead of immediately failing, improving the user experience when retrying cleared tasks.

Key Changes:

• Introduced poke_retry_number property to distinguish between initial sensor attempts and subsequent pokes
• Added _store_producer_task_state callback to track producer task failures via XCom
• Enhanced consumer sensor logic to fallback to local execution for upstream-failed tasks
• Added comprehensive test coverage for the new failure handling scenarios

Reviewed Changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 3 comments.

File	Description
cosmos/operators/watcher.py	Added producer state tracking via XCom and upstream failure detection logic in consumer sensor
tests/operators/test_watcher.py	Added tests for producer state handling, callback management, and upstream failure scenarios
docs/getting_started/watcher-execution-mode.rst	New documentation file explaining the WATCHER execution mode feature
cosmos/init.py	Version bump to 1.11.0a13

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 97.77%. Comparing base (135147f) to head (cd48ef8).
⚠️ Report is 3 commits behind head on main.

Additional details and impacted files

@@           Coverage Diff           @@
##             main    #2062   +/-   ##
=======================================
  Coverage   97.76%   97.77%           
=======================================
  Files          91       91           
  Lines        5824     5830    +6     
=======================================
+ Hits         5694     5700    +6     
  Misses        130      130           

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.