Refactor Watcher InvocationMode.SUBPROCESS log parser

[tatiana]

While the PR #2152 introduced the capability of users running the ExecutionMode.WATCHER with dbt installed in a separate virtualenv, and still get updates in "real time", it came with two downsides:

1. Spread the watcher logic into multiple areas of the Cosmos code base (inside the utils folder, and also inside the hooks/subprocess - which should be focused only on running suprocess);
2. It changed the subprocess hook to always invoke the method _store_dbt_resource_status_from_log(line, **kwargs) for each log line made available in stdout, even when users are not using ExecutionMode.WATCHER .

This PR refactors the implementation to keep watcher implementation details inside the operators/ folder. It also allows operators that subclass the AbstractDbtLocalBase to define the attribute _process_log_line_callable. Only classes that define this function will parse the log lines with the desired logic. At the moment, only the DbtProducerWatcherOperator implements the logic of uploading the status to XCom.

[Copilot]

Pull request overview

This PR refactors the InvocationMode.SUBPROCESS log parser implementation for the Watcher execution mode to improve code organization and reduce unnecessary processing overhead. The main motivation is to consolidate watcher-specific logic into the operators folder and avoid parsing logs when not using the Watcher execution mode.

Key Changes:

• Moved _store_dbt_resource_status_from_log from the subprocess hook to the watcher operator module
• Introduced _process_log_line_callable attribute in AbstractDbtLocalBase to allow selective log line processing
• Relocated watcher state utilities from _utils/watcher_state.py to operators/_watcher/state.py

Reviewed changes

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

Show a summary per file

File	Description
cosmos/hooks/subprocess.py	Removed log parsing method and added optional process_log_line parameter to delegate log processing to callers
cosmos/operators/local.py	Added _process_log_line_callable attribute and wired it to the subprocess hook's run_command method
cosmos/operators/watcher.py	Implemented _store_dbt_resource_status_from_log as a standalone function and set it as the callable for DbtProducerWatcherOperator
cosmos/operators/_watcher/init.py	Created new module to expose watcher state utilities
cosmos/_triggers/watcher.py	Updated import path for build_producer_state_fetcher
tests/hooks/test_subprocess.py	Updated tests to call the relocated function and adjusted mock paths

---

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

[codecov]

Codecov Report

❌ Patch coverage is 96.15385% with 1 line in your changes missing coverage. Please review.
✅ Project coverage is 97.84%. Comparing base (e6ee9a2) to head (d215d6f).
⚠️ Report is 3 commits behind head on main.

Files with missing lines	Patch %	Lines
cosmos/hooks/subprocess.py	66.66%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2183      +/-   ##
==========================================
- Coverage   97.86%   97.84%   -0.02%     
==========================================
  Files          93       94       +1     
  Lines        6031     6036       +5     
==========================================
+ Hits         5902     5906       +4     
- 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.

[pankajastro]

Looks good. I have posted a minor comment.

[tatiana]

It seems the only coverage concern is a line that was already not covered before this PR, and was highlighted due to the file rename:
https://app.codecov.io/gh/astronomer/astronomer-cosmos/pull/2183/blob/cosmos/operators/_watcher/state.py?dropdown=coverage

Therefore, I'm going ahead and merging the change.