Fix: watcher consumer retry logging dbt with --log-format json

[tatiana]

When a watcher consumer task retries via _fallback_to_non_watcher_run, it copies the producer's add_cmd_flags() output into its own dbt invocation. The producer hardcodes log_format="json" so it can parse dbt's structured event stream — but that internal choice was leaking into the consumer's user-facing retry dbt command, producing raw JSON (one structured event per line) instead of dbt's default formatted text in task logs.

Example of what users currently see on a retry task like watcher_downstream_not_skipped.model_downstream_run:

[2026-05-20 21:45:11] INFO - {"data": {"log_version": 3, "version": "=1.11.8"}, "info": {"category": "", "code": "A001", ...}}
[2026-05-20 21:45:11] INFO - {"data": {"adapter_name": "***", ...}, "info": {"category": "", "code": "E034", ...}}

Fix

Add --log-format to _filter_flags's skip list. The producer's hardcoded --log-format json is no longer propagated into the consumer's fallback dbt command, so retries default to dbt's normal text output.

How users can still get JSON

Users who want JSON-formatted dbt output on retry can opt in via:

DbtDag(
    ...,
    operator_args={"dbt_cmd_flags": ["--log-format", "json"]},
)

dbt_cmd_flags is appended to the dbt command by build_cmd outside of _filter_flags's pipeline, so this path keeps working after the fix.

Test plan

Automated

• tests/operators/test_watcher.py updated:
  • test_filter_flags — extended to assert --log-format <value> is filtered alongside --select / --exclude
  • test_filter_flags_strips_log_format_when_first (new) — covers the edge case where --log-format is the first flag in the input list
  • test_fallback_strips_producer_log_format_by_default (new) — end-to-end verifies the consumer's retry command has no --log-format
  • test_fallback_preserves_user_dbt_cmd_flags_opt_in_to_json (new) — documents the opt-in path via operator_args["dbt_cmd_flags"]
• All 12 affected tests pass on tests.py3.12-2.11-1.11.

Manual validation DAGs

Two example DAGs in dev/failed_dags/ exercise both sides of the change. Both intentionally fail once on model_retry and retry via the watcher fallback path — inspect the retry task's logs to verify behaviour. Do not run them concurrently (they share public._cosmos_fail_once_seq).

• dev/failed_dags/example_watcher_downstream_not_skipped.py — default config (no dbt_cmd_flags). On the watcher_downstream_not_skipped.model_retry_run retry, log lines should be dbt's normal text format (e.g. INFO [MainThread]: Running with dbt=1.11.8). Before this fix, those lines came through as raw JSON; after the fix, they should read normally.
• dev/failed_dags/example_watcher_downstream_not_skipped_json_logs.py (new in this PR) — adds operator_args={"dbt_cmd_flags": ["--log-format", "json"]}. On the watcher_downstream_not_skipped_json_logs.model_retry_run retry, log lines should be structured JSON objects ({"data": ..., "info": ...}). Confirms the user opt-in still works after the fix.

How this was discovered

Reported while validating example_watcher_downstream_not_skipped's model_downstream_run retry on the 1.14.2a3 release candidate.

Backport

Targeting main; needs a backport to release-1.14 so it ships in 1.14.2. Add Cosmos 1.14.2 milestone after merge.

[Copilot]

Pull request overview

This PR prevents the watcher producer’s internal --log-format json setting (needed for structured event parsing) from leaking into the watcher consumer’s fallback retry dbt invocation, so retry task logs default back to dbt’s normal human-readable text output.

Changes:

• Extend the watcher consumer’s retry flag filtering to drop --log-format <value> alongside --select/--exclude.
• Clarify docstrings around why --log-format is filtered and how users can still opt into JSON via operator_args["dbt_cmd_flags"].
• Add/extend tests covering both the filtering behavior and the end-to-end fallback command construction.

Reviewed changes

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

File	Description
cosmos/operators/_watcher/base.py	Adds --log-format to the _filter_flags skip list and updates retry/filter documentation.
tests/operators/test_watcher.py	Adds new regression tests to ensure producer --log-format json is stripped on fallback and user opt-in via dbt_cmd_flags remains intact.

---

💡 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.04%. Comparing base (f8087c5) to head (595e87a).

Additional details and impacted files

@@           Coverage Diff           @@
##             main    #2713   +/-   ##
=======================================
  Coverage   98.04%   98.04%           
=======================================
  Files         105      105           
  Lines        7892     7892           
=======================================
  Hits         7738     7738           
  Misses        154      154           

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

[pankajkoti]

LGTM. I guess we need to associate the milestone 1.14.2 for this PR?

[tatiana]

🚀 Released in Cosmos 1.14.2 (PyPI).