Add deferrable support for Watcher Sensor

[pankajastro]

This PR adds async/deferrable support to the Watcher sensor

Benefits

• Reduces resource usage for long-running watcher tasks.
• Improves task scheduling and throughput.

Limitation

• Supports deferrable execution for model or run commands only.
• Retry execution remains synchronous, and deferrable happens only for try no 0.

Future

• Improve poke_interval and timeout handling to address performance issue: Support operator_args for watcher tasks #1973

closes: #2059

[codecov]

Codecov Report

❌ Patch coverage is 98.87640% with 1 line in your changes missing coverage. Please review.
✅ Project coverage is 97.81%. Comparing base (a95c81b) to head (288274d).
⚠️ Report is 4 commits behind head on main.

Files with missing lines	Patch %	Lines
cosmos/_triggers/watcher.py	98.57%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2084      +/-   ##
==========================================
+ Coverage   97.80%   97.81%   +0.01%     
==========================================
  Files          91       92       +1     
  Lines        5871     5948      +77     
==========================================
+ Hits         5742     5818      +76     
- Misses        129      130       +1     

☔ 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

This PR adds deferrable (async) execution support to the DbtConsumerWatcherSensor by introducing a new WatcherTrigger class. When the sensor's initial poke returns False, it defers execution to the trigger which asynchronously polls XCom for model status updates.

• Implemented WatcherTrigger class for async polling of dbt model status
• Added execute and execute_complete methods to DbtConsumerWatcherSensor for deferrable behavior
• Refactored XCom decompression logic into a reusable _parse_compressed_xcom utility function
• Added comprehensive test coverage for the trigger and sensor deferral behavior

Reviewed Changes

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

Show a summary per file

File	Description
cosmos/_triggers/watcher.py	New trigger class for async polling of dbt model status from XCom
cosmos/_triggers/init.py	Empty init file for the _triggers module
cosmos/operators/watcher.py	Added execute/execute_complete methods and refactored to use shared decompression utility
tests/_triggers/watcher.py	Comprehensive test coverage for WatcherTrigger async behavior
tests/_triggers/init.py	Empty init file for test triggers module
tests/operators/test_watcher.py	Tests for sensor deferral and execute_complete method
pyproject.toml	Added pytest-asyncio dependency for async test support

---

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

[pankajkoti]

It's looking great to me. Have a couple of comments inline.

Also, looks like a few tests are failing in the CI.

[tatiana]

@pankajastro, this is really exciting, thanks for finding a solution for it!
Do you see any downsides or risks with always using the deferrable implementation? What happens if the end-user does not have a triggerer in their Airflow deployment?
Should we also support people opting out of using the deferrable mode?
Finally, are there any performance improvements when switching to deferrable, when running ExecutionMode.WATCHER with our benchmark project?

[pankajastro]

Do you see any downsides or risks with always using the deferrable implementation? What happens if the end-user does not have a triggerer in their Airflow deployment?

I’m a bit concerned about the number of database accesses from the trigger.

Should we also support people opting out of using the deferrable mode?

I’ve thought about it, and I think we should expose a parameter at the sensor level to conditionally defer. However, it would be better to do this once #1973 is solved.

Finally, are there any performance improvements when switching to deferrable, when running ExecutionMode.WATCHER with our benchmark project?

I have not done any performance testing.

[tatiana]

Looks great, thank you very much, @pankajastro !