Support real-time consumer updates when using ExecutionMode.WATCHER and InvocationMode.SUBPROCESS

[pankajastro]

closes: #2060
This PR modifies ExecutionMode.WATCHER so that, when InvocationMode.SUBPROCESS is used, sensor tasks can still be updated in real time based on dbt status information from structured JSON logs, rather than parsing run_results.json after the full dbt pipeline is executed. The change enables real-time status tracking for subprocess-based dbt invocations.

Key Changes:

• Added --log-format json flag to dbt build commands in watcher mode
• Implemented real-time JSON log parsing in subprocess hook to extract and store node status to XCom
• Refactored XCom utility functions to a new common module for better code organization

We can see here that the producer task is running but some node has already succeeded for WATCHER mode with SUBPROCESS invocation

[Copilot]

Pull request overview

This PR modifies the WATCHER execution mode and SUBPROCESS invocation mode to consume dbt status information from structured JSON logs rather than parsing run_results.json after execution completes. The change enables real-time status tracking for subprocess-based dbt invocations.

Key Changes:

• Added --log-format json flag to dbt build commands in watcher mode
• Implemented real-time JSON log parsing in subprocess hook to extract and store node status to XCom
• Refactored XCom utility functions to a new common module for better code organization

Reviewed changes

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

Show a summary per file

File	Description
cosmos/operators/watcher.py	Removed local safe_xcom_push (moved to utils), added log_format="json" configuration, switched from parsing run_results to direct XCom status lookup
cosmos/operators/local.py	Added **kwargs parameter support to pass context through subprocess/runner invocations
cosmos/operators/base.py	Added log_format parameter to DbtBuildMixin to support JSON log output configuration
cosmos/hooks/subprocess.py	Implemented _store_dbt_resource_status_from_log to parse JSON logs, changed subprocess to text mode, stores node status to XCom in real-time
cosmos/dbt/runner.py	Added **kwargs parameter to support additional arguments
cosmos/_utils/common.py	New utility module containing safe_xcom_push and get_xcom_val functions
cosmos/_triggers/watcher.py	Updated _parse_node_status to handle new status key format for non-event mode
cosmos/init.py	Version bump to 1.12.0a3

Comments suppressed due to low confidence (1)

cosmos/hooks/subprocess.py:80

• The docstring for run_command should be updated to document the new **kwargs parameter. Since kwargs is now used to pass the context (which contains the TaskInstance for XCom operations), this should be documented. Add:

:param kwargs: Additional keyword arguments. Can include 'context' for XCom operations.

    def run_command(
        self,
        command: list[str],
        env: dict[str, str] | None = None,
        output_encoding: str = "utf-8",
        cwd: str | None = None,
        **kwargs: Any,
    ) -> FullOutputSubprocessResult:
        """
        Execute the command.

        If ``cwd`` is None, execute the command in a temporary directory which will be cleaned afterwards.
        If ``env`` is not supplied, ``os.environ`` is passed

        :param command: the command to run
        :param env: Optional dict containing environment variables to be made available to the shell
            environment in which ``command`` will be executed.  If omitted, ``os.environ`` will be used.
            Note, that in case you have Sentry configured, original variables from the environment
            will also be passed to the subprocess with ``SUBPROCESS_`` prefix.
        :param output_encoding: encoding to use for decoding stdout
        :param cwd: Working directory to run the command in.
            If None (default), the command is run in a temporary directory.
        :return: :class:`namedtuple` containing:
                                    ``exit_code``
                                    ``output``: the last line from stderr or stdout
                                    ``full_output``: all lines from stderr or stdout.
        """

---

💡 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 97.86%. Comparing base (4bd8e9e) to head (39d3609).
⚠️ Report is 1 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2152      +/-   ##
==========================================
+ Coverage   97.81%   97.86%   +0.04%     
==========================================
  Files          93       93              
  Lines        6005     6031      +26     
==========================================
+ Hits         5874     5902      +28     
+ Misses        131      129       -2     

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

[tatiana]

This looks very good, @pankajastro. Thanks a lot for the work.
I left minor feedback, and I'm happy for the changes to be merged.