Add DbtProducerWatcherOperator for the proposed ExecutionMode.WATCHER

[pankajkoti]

This PR introduces the DbtProducerWatcherOperator in
cosmos/operators/watcher.py for use with the proposed ExecutionMode.WATCHER.
The operator triggers a single dbt build and streams real-time per-model
run statuses via dbtRunner events, pushing keys such as nodefinished_<uid>,
aggregated dbt_startup_events, and falling back to pushing run_results
from the target directory to XCom when dbtRunner is unavailable.

Additionally, local execution has been updated with a
_push_run_results_to_xcom helper and a push_run_results_to_xcom flag,
enabling gzip+base64–compressed run-results to be stored in XCom for
fallback support.

closes: #1958
closes: https://github.com/astronomer/oss-integrations-private/issues/238

[netlify]

✅ Deploy Preview for sunny-pastelito-5ecb04 canceled.

Name	Link
🔨 Latest commit	3f25593
🔍 Latest deploy log	https://app.netlify.com/projects/sunny-pastelito-5ecb04/deploys/68d50c0ea796e40008047a44

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 97.77%. Comparing base (8627d63) to head (dd32126).

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #1982      +/-   ##
==========================================
+ Coverage   97.67%   97.77%   +0.10%     
==========================================
  Files          87       87              
  Lines        5371     5443      +72     
==========================================
+ Hits         5246     5322      +76     
+ Misses        125      121       -4     

☔ 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 introduces a new DbtBuildCoordinatorOperator that enables real-time monitoring of dbt model execution status through XCom for the proposed ExecutionMode.WATCHER. The operator provides two execution paths: streaming mode using dbtRunner events for real-time per-model status updates, and fallback mode that pushes compressed run results after completion.

Key changes:

• New DbtBuildCoordinatorOperator class with streaming and fallback execution modes
• Support for pushing compressed run results to XCom via new push_run_results_to_xcom parameter
• Enhanced local execution operators to support XCom-based result sharing

Reviewed Changes

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

File	Description
cosmos/operators/watcher.py	Implements the main DbtBuildCoordinatorOperator with event streaming and XCom coordination
cosmos/operators/local.py	Adds _push_run_results_to_xcom helper method and push_run_results_to_xcom parameter support
cosmos/operators/virtualenv.py	Updates run_command method to pass through the new push_run_results_to_xcom parameter
tests/operators/test_watcher.py	Comprehensive test suite for the new operator functionality

Comments suppressed due to low confidence (1)

cosmos/operators/local.py:1

• Remove commented-out code. If this alternative implementation is needed for future reference, document the reasoning or move it to documentation.

from __future__ import annotations

---

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

[tatiana]

Looks great, @pankajkoti , thanks a lot for the great work and addressing the feedback.
I'm happy for us to merge this PR once the follow up ticket is logged.