Fix duplicate log lines in watcher subprocess execution and format timestamps

[pankajkoti]

Changes:

• Remove duplicate logging of dbt messages in store_dbt_resource_status_from_log(). The message was being logged twice: once in the else block and again at the end of the function. Now it's only logged once with the appropriate log level.

• Format timestamps to match dbt runner format (HH:MM:SS) instead of full ISO format. This provides consistent log output between subprocess and dbt_runner modes. Example: "13:16:05 Running with dbt=1.10.11"

• Add unit tests to verify:

  • Log messages are logged exactly once (no duplicates)
  • Timestamps are formatted correctly as HH:MM:SS

• Extend integration tests (test_dbt_dag_with_watcher, test_dbt_dag_with_watcher_and_subprocess) to verify no duplicate log messages in both dbt_runner and subprocess modes

related: #2287
related: #2241

[netlify]

✅ Deploy Preview for astronomer-cosmos canceled.

Name	Link
🔨 Latest commit	e6d35bf
🔍 Latest deploy log	https://app.netlify.com/projects/astronomer-cosmos/deploys/697b8a522c9d600008a3d085

[tatiana]

Thanks a lot, @pankajkoti. It was probably my mistake during rebasing.

Looks great! I do not believe we need timestamp parity. From my perspective, this makes our codebase more complex without any clear benefit, because we can already see the Airflow timestamp when we're invoking from Airflow. If there was an easy way, I'd remove that timestamp from the dbtRunner implementation 😂

That said, I believe the duplicate issue is the biggest problem and I'm happy for things to be the same, if you prefer - we'll just need to add tests to cover the additional logic.

[pankajkoti]

Previously, we were seeing duplicate logs in the Watcher Subprocess execution

Now, after the change in this PR, we have only one log, and it also includes the timestamp to be consistent with the dbt Runner logs output

Subprocess output:

dbt Runner output:

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 98.07%. Comparing base (6fabe60) to head (e6d35bf).
⚠️ Report is 3 commits behind head on main.

Additional details and impacted files

@@           Coverage Diff           @@
##             main    #2301   +/-   ##
=======================================
  Coverage   98.07%   98.07%           
=======================================
  Files         100      100           
  Lines        6452     6458    +6     
=======================================
+ Hits         6328     6334    +6     
  Misses        124      124           

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

Looks great! I do not believe we need timestamp parity. From my perspective, this makes our codebase more complex without any clear benefit, because we can already see the Airflow timestamp when we're invoking from Airflow. If there was an easy way, I'd remove that timestamp from the dbtRunner implementation 😂

That said, I believe the duplicate issue is the biggest problem and I'm happy for things to be the same, if you prefer - we'll just need to add tests to cover the additional logic.

Yes, I am a bit torn on this one. I am slightly inclined to preserve the timestamp for the following reasons:

1. Only watcher subprocess logs do not have a timestamp value. Other execution modes' subprocess runs still include the timestamp
2. This timestamp is the real timestamp given by dbt event. Airflow logs might have slight delays (at times, maybe?) and reorderings of logs are possible at a time, based on how the flushing works(?).
3. If users have a regex configured for filtering/processing logs forwarded to their log store, the consistency in logs could still benefit, I believe.

Hence, I am thinking of preserving this for now. But if we feel it is a maintenance burden happy for it to be removed in a subsequent PR.

[Copilot]

Pull request overview

This PR fixes duplicate log lines in watcher subprocess execution and standardizes timestamp formatting. The changes ensure that dbt log messages are logged exactly once (previously logged twice due to duplicate logic) and format timestamps in HH:MM:SS format to match dbt runner output instead of full ISO format.

Changes:

• Removed duplicate logging in store_dbt_resource_status_from_log() function
• Added timestamp formatting to convert ISO format to HH:MM:SS format
• Added comprehensive unit and integration tests to verify single logging and timestamp formatting

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	Removed duplicate log statement and added timestamp formatting logic with error handling
tests/operators/test_watcher.py	Added unit tests for single logging, timestamp formatting, and invalid timestamp handling; extended integration tests to verify no duplicates

---

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