Keep watcher sensor polling when producer is still running

[pankajkoti]

Previously, any retry of a watcher consumer sensor (try_number > 1) unconditionally fell back to a non-watcher dbt run. This meant that if the sensor timed out while the producer was still working, the retry would launch a duplicate dbt invocation instead of continuing to wait for the producer's XCom updates.

On retry, first check the producer task state. If the producer has reached a terminal state (success, failed, skipped, upstream_failed, removed), fall back to a non-watcher run as before — this covers both producer completion and manual task clears from the UI. Otherwise, keep polling so the sensor can pick up where it left off without spawning a redundant run.

closes: https://github.com/astronomer/oss-integrations-private/issues/359

[pankajkoti]

Retry attempt on the sensor when the producer is still running:

Successful completion of DAG afterwards:

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 98.05%. Comparing base (e9a43fc) to head (b33fe56).
⚠️ Report is 1 commits behind head on main.

Additional details and impacted files

@@           Coverage Diff           @@
##             main    #2592   +/-   ##
=======================================
  Coverage   98.04%   98.05%           
=======================================
  Files         104      104           
  Lines        7737     7747   +10     
=======================================
+ Hits         7586     7596   +10     
  Misses        151      151           

☔ 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.

[Copilot]

Pull request overview

Adjusts watcher consumer sensor retry behavior to avoid spawning duplicate dbt runs when the watcher producer is still running, by checking the producer task state and only falling back to a non-watcher run when the producer is in a terminal state.

Changes:

• Add producer-terminal state detection (PRODUCER_TERMINAL_STATES, is_producer_task_terminated) and use it during consumer-sensor retries.
• Update watcher consumer sensor retry flow to keep polling when the producer is still active, otherwise fall back to non-watcher execution.
• Extend unit tests to cover “retry + producer running” vs “retry + producer terminated” scenarios (local and Kubernetes).

Reviewed changes

Copilot reviewed 6 out of 6 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
cosmos/operators/_watcher/base.py	Implements retry handler that conditionally falls back or continues polling based on producer task state.
cosmos/operators/_watcher/state.py	Defines producer terminal states and helper to classify producer task termination.
tests/operators/_watcher/test_state.py	Adds unit tests for the new producer-terminal-state helper.
tests/operators/test_watcher.py	Updates/expands watcher sensor tests to validate retry behavior for running vs terminated producer.
tests/operators/test_watcher_kubernetes_unit.py	Adds Kubernetes watcher sensor retry test to ensure polling continues when producer is running.
dev/dags/cosmos_manifest_selectors_example.py	Adjusts example selector list to include additional upstream sources.

---

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

[tatiana]

Looks great, thanks a lot, @pankajkoti for creating a new PR and rebasing after the many changes introduced in the meantime.