Simplify and unify WATCHER implementation regardless of InvocationMode

[tatiana]

This PR unifies ExecutionMode.WATCHER so it works almost identically with both InvocationMode.SUBPROCESS and InvocationMode.DBT_RUNNER - removing the previously separate code paths that depended on the invocation mode.

Problem

Before this PR, ExecutionMode.WATCHER had two completely different strategies depending on how dbt was invoked:

• InvocationMode.DBT_RUNNER: Subscribed to dbt's EventMsg callbacks, serialised each NodeFinished event as zlib-compressed base64 JSON under XCom key nodefinished_<uid>, and
  pushed startup metadata under dbt_startup_events.
• InvocationMode.SUBPROCESS: After dbt exited, read target/run_results.json and pushed the entire results blob once under XCom key run_results. Sensors could not react per-model
  until the full build finished.

	InvocationMode.DBT_RUNNER	InvocationMode.SUBPROCESS
Producer mechanism	_dbt_runner_callbacks with structured EventMsg objects	_process_log_line_callable + --log-format json parsing
XCom key (status)	nodefinished_{uid} (compressed+base64 JSON)	{uid}_status (plain string)
Consumer path	_get_status_from_events() (decompresses JSON)	get_xcom_val(…_status)
Trigger path	_get_node_status() → decompresses JSON	_get_node_status() → reads plain value

This meant the consumer sensors needed to know the invocation mode at poke time (use_event() method), and the triggerer needed a use_event flag. The code was complex, duplicated, and the two modes delivered very different latency and observability.

Solution

Both invocation modes now feed through the same store_dbt_resource_status_from_log parser, which pushes per-model XCom keys <unique_id>_status as each NodeFinished event arrives:

• InvocatioMode.SUBPROCESS: Each JSON log line from stdout is parsed directly by the callable wired to run_subprocess. `
• InvocationMode.DBT_RUNNER: Each EventMsg from the dbt callback is serialised to JSON via google.protobuf.json_format.MessageToJson (a transitive dbt-core dependency) and passed
  through the same parser. Stdout is redirected to suppress raw JSON duplication in Airflow logs.

Consumer sensors and the WatcherTrigger always poll the canonical {uid}_status key regardless of invocation mode.

Key changes

cosmos/operators/watcher.py — DbtProducerWatcherOperator

• Added run_dbt_runner() override: registers a MessageToJson-based callback that feeds store_dbt_resource_status_from_log. Callback is only registered when context
  is not None (guards against dbt deps calls during _install_dependencies, which would otherwise cause a phantom callback and GenericExceptionOnRun errors).
• Added contextlib.redirect_stdout(io.StringIO()) during the main build to suppress raw JSON from dbt's stdout (since events are handled via the callback).
• Removed EventMsg-based _handle_node_finished / _serialize_event / zlib-compressed XCom strategy.
• Removed InvocationMode-based branching from execute().

cosmos/operators/_watcher/base.py

• Removed use_event() abstract method and _get_status_from_events() / _get_status_from_run_results() — no longer needed.
• _get_node_status() now always reads {uid}_status XCom key (plus the _tests_status key for DbtTestWatcherOperator).
• poke() calls _log_startup_events() for all non-test sensors regardless of invocation mode.
• _process_dbt_log_event() simplified: removed the EventMsg branch (dict-only now).
• Added _log_dbt_msg() helper to improve log UX: shows human-readable dbt messages (e.g. "Found 5 models, 3 seeds...") instead of raw JSON in Airflow task logs.
  Reduces cyclomatic complexity of store_dbt_resource_status_from_log from 11 to 9 (fixes Ruff C901).
• assert context is not None replaced with if context is None: return — avoids AssertionError("") → GenericExceptionOnRun when dbt's phantom callback fires without
  a context.
• _log_startup_events() now checks isinstance(dbt_startup_events, list) before iterating.

cosmos/operators/_watcher/triggerer.py

• Removed use_event parameter from WatcherTrigger.
• _parse_dbt_node_status_and_compiled_sql() simplified: always reads {uid}_status (no nodefinished_* key lookup).

cosmos/operators/watcher_kubernetes.py

• Removed use_event() stub.

Tests

• Removed tests for deleted code paths (_handle_node_finished, _get_status_from_events, _get_status_from_run_results, use_event()).
• Switched test_dbt_dag_with_watcher from capsys to caplog to avoid thread-safety issues with dbt's multi-threaded stdout writes.
• Fixed test_dbt_runner_caching_and_callbacks to remove the now-unnecessary EventMsg patch.

Test plan

• Unit tests: hatch run tests.py3.11-2.10-1.9:test
• Integration tests verify ExecutionMode.WATCHER runs end-to-end with both SUBPROCESS and DBT_RUNNER invocation modes
• airflow dags test example_watcher completes without getting stuck or emitting GenericExceptionOnRun errors
• Trigger watcher DAGs with airflow standalone and validate the output.

[Copilot]

Pull request overview

This PR unifies ExecutionMode.WATCHER behavior across InvocationMode.SUBPROCESS and InvocationMode.DBT_RUNNER by routing both execution paths through the same dbt JSON log parsing pipeline and standardizing per-node XCom keys.

Changes:

• Standardize per-node status reporting to <unique_id>_status for both invocation modes via store_dbt_resource_status_from_log.
• Replace dbt Runner event-specific XCom compression strategy with an EventMsg→JSON→parser callback.
• Simplify consumer sensors + triggerer to always poll the canonical *_status (and *_tests_status) keys; update/trim tests accordingly.

Reviewed changes

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

Show a summary per file

File	Description
cosmos/operators/watcher.py	Producer now always uses --log-format json; adds dbt Runner callback that feeds the unified parser; removes event-compression path.
cosmos/operators/_watcher/base.py	Simplifies event processing to dict-only; standardizes _get_node_status() to *_status; adds _log_dbt_msg(); makes startup-events logging safer.
cosmos/operators/_watcher/triggerer.py	Removes use_event branching; trigger always reads *_status; removes compressed-event parsing helper.
cosmos/operators/_watcher/__init__.py	Stops exporting _parse_compressed_xcom after triggerer simplification.
cosmos/operators/watcher_kubernetes.py	Removes use_event() stub now that event-vs-xcom branching is gone.
tests/operators/test_watcher.py	Updates tests to new canonical XCom strategy; adds coverage for new runner callback wiring and startup-event logging; switches capsys→caplog.
tests/operators/test_watcher_kubernetes_unit.py	Removes use_event() test (method deleted).
tests/operators/_watcher/test_watcher_base.py	Removes abstract-method tests for deleted use_event()/event-based status methods.
tests/operators/_watcher/test_triggerer.py	Updates trigger serialization/tests for removed use_event and canonical *_status lookup.
tests/hooks/test_subprocess.py	Updates expectations for store_dbt_resource_status_from_log to no longer assert on missing context.
tests/dbt/test_runner.py	Removes no-longer-needed EventMsg patching.

---

💡 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 98.01%. Comparing base (3f7b3fa) to head (8834e50).
⚠️ Report is 7 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2498      +/-   ##
==========================================
- Coverage   98.03%   98.01%   -0.02%     
==========================================
  Files         103      103              
  Lines        7262     7164      -98     
==========================================
- Hits         7119     7022      -97     
+ Misses        143      142       -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

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

---

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

[pankajkoti]

Looks like a solid refactor to me! Thanks for adding all the code comments, making the review easier and the code maintainable.

I did a sanity test on both SUBPROCESS and dbt Runner approaches, and it's looking fine to me with respect to pushing XCOMs, creating asset events, populating compiled SQLs, consistent logs and dag run durations. Also, the thread count was set to 12 in my BigQuery profile while testing, so it's looking concurrency safe. Good net 400+ lines being removed :)

[michal-mrazek]

Looking good, I tested in local docker compose. Just added one small suggestion - it took me some time to figure out that the if context is None is just typing thing :)

[Copilot]

Copilot reviewed 11 out of 11 changed files in this pull request and generated no new comments.

---

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

[Copilot]

Copilot reviewed 11 out of 11 changed files in this pull request and generated no new comments.

---

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