Guard store_dbt_resource_status_from_log against falsy extra_kwargs

The refactor dropped the original `if extra_kwargs` guard, leaving several
unconditional `extra_kwargs.get(...)` calls that would raise AttributeError if
extra_kwargs were None/non-mapping. Normalise `extra_kwargs = extra_kwargs or {}`
at the top so all accesses are safe, restoring the original falsy-handling.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: pankajastro

cosmos/operators/_watcher/base.py

@@ -469,6 +469,8 @@ def store_dbt_resource_status_from_log(
         terminal state and may be re-flushed if a later error event provides the message (e.g. subprocess mode).
         This replaces the previous per-event XCom push, which wrote on every event (lock contention) and let a trailing ``NodeFinished`` overwrite the captured error message.
     """
+    # extra_kwargs is typed Any and may be falsy/None; normalise so the .get() calls below are safe.
+    extra_kwargs = extra_kwargs or {}
     try:
         log_line = json.loads(line)
     except json.JSONDecodeError:

tests/operators/_watcher/test_watcher_base.py

@@ -9,6 +9,7 @@
     _flush_dbt_event,
     _flush_dbt_event_on_error,
     _process_dbt_log_event,
+    store_dbt_resource_status_from_log,
 )
 from cosmos.operators.local import DbtRunLocalOperator
 
@@ -351,6 +352,13 @@ def test_none_string_status_is_ignored(self):
         _accumulate_dbt_log_event(buffer, self._run_result_error(msg="err"))  # node_status == "None"
         assert buffer[self.UID]["status"] is None
 
+    @pytest.mark.parametrize("extra_kwargs", [None, {}])
+    def test_store_dbt_resource_status_from_log_tolerates_falsy_extra_kwargs(self, extra_kwargs):
+        """A falsy/None extra_kwargs must not raise during parsing (no context to push to)."""
+        line = '{"info": {"name": "NodeFinished"}, "data": {"node_info": {"unique_id": "model.p.m", "node_status": "success"}}}'
+        # Should be a no-op (no context/ti) rather than raising AttributeError.
+        store_dbt_resource_status_from_log(line, extra_kwargs, node_event_buffer={})
+
     def test_accumulate_ignores_non_allowlisted_and_missing_unique_id(self):
         buffer: dict = {}
         _accumulate_dbt_log_event(