Prevent watcher producers from colliding on one XCom-backup key

[tatiana]

Closes #2625.
Closes BOSS-83.
Closes BOSS-157.

The problem

When a DAG contained multiple DbtTaskGroups using ExecutionMode.WATCHER, every dbt_producer_watcher in the same DAG run computed the same Airflow Variable key for its XCom backup:

cosmos_xcom_backup__{dag_id}__{run_id}

The task-group segment was supposed to discriminate concurrent producers, but the helper that read it (_get_task_group_id in cosmos/operators/_watcher/xcom.py) was reading task.task_group_id — an attribute that does not exist on Airflow operators in either Airflow 2 or Airflow 3 (the canonical path is task.task_group.group_id). Because the read went through getattr(task, "task_group_id", None), it silently returned None; the segment was omitted from the key, and every producer in the DAG run wrote to the same Variable row.

The consequence depends on the Airflow version:

• Airflow 2: the second producer to insert raises UniqueViolation on variable_key_uq — a noisy ERROR on every model completion (this is the original symptom from the issue report).
• Airflow 3: Variable.set is an upsert (ON CONFLICT DO UPDATE), so the conflict is silently absorbed and the last writer wins. If one producer fails and retries, _restore_xcom_from_variable may load the other group's XCom snapshot — an undetected correctness bug.

How to reproduce

dev/failed_dags/example_watcher_xcom_collision.py reproduces the issue end-to-end against the jaffle_shop project: two DbtTaskGroups (group_a selecting +customers, group_b selecting +orders), no inter-group dependency, both running in parallel.

Reproduction conditions (from the issue report):

• both groups start in the same DAG run (no dependency between them → parallel)
• each group must have at least one model complete while both producers run
• Airflow must be running with multiple workers (not LocalExecutor with parallelism=1)

Trigger the DAG manually; on the first node-status XCom in whichever producer loses the INSERT race, you see — before this fix:

• on Airflow 2:

  {watcher.py:268} ERROR - Error in dbt event callback:
  (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "variable_key_uq"
  DETAIL: Key (key)=(cosmos_xcom_backup__example_watcher_xcom_collision__manual__...) already exists.
  

• on Airflow 3: nothing in the logs (silent overwrite of the backup row).

After this fix the two producers compute distinct keys (…__group_a__… and …__group_b__…) and neither symptom occurs.

The fix

_get_task_group_id now reads task.task_group.group_id — the same path that _get_dbt_dag_task_group_identifier in cosmos/airflow/graph.py already uses. .task_group/.group_id are accessed directly (not via getattr with a None default), so a future rename or typo at this site raises AttributeError at first invocation instead of silently degrading to the colliding-key behaviour. The outer getattr(ti, "task", None) is kept because the runtime-TI shape varies legitimately between TaskInstance, RuntimeTaskInstance, and test stubs.

How do we prevent this from happening again

Three independent reasons let the bug slip through, addressed in three independent ways:

1. The unit test masked the bug with MagicMock. The previous test did ti.task = MagicMock(task_group_id="my_group"). Because MagicMock fabricates any attribute access, getattr(mock, "task_group_id", None) returned "my_group" and the test passed — even though the same call against a real operator always returned None. The test (tests/operators/_watcher/test_state.py::TestInitXcomBackup) is reworked to mock task.task_group.group_id, and a new test test_rejects_task_without_task_group_attribute uses MagicMock(spec=[...]) to forbid fabrication and asserts the helper raises loudly when fed a malformed task.

2. No test asserted uniqueness across task groups. A new integration test test_dbt_task_groups_with_watcher_produce_distinct_xcom_backup_keys in tests/operators/test_watcher.py builds two DbtTaskGroups in one DAG, drives _init_xcom_backup against each real producer operator (exercising the actual code path the bug lived in), and asserts the resulting backup keys differ — and contain group_a/group_b. This test fails on main with the same colliding key the user sees in production, and passes after the helper fix.

3. Direct attribute access replaces silent-fallback getattr on the contract attributes (task.task_group, .group_id). A future typo or rename at the failure site now surfaces as AttributeError at first invocation instead of degrading to a colliding key. This is the safer failure mode the original getattr chain lacked.

🤖 Generated with Claude Code

[Copilot]

Pull request overview

Fixes issue #2625 where multiple DbtTaskGroups using ExecutionMode.WATCHER in the same DAG collide on a single XCom backup Variable key. The root cause is _get_task_group_id reading the nonexistent task.task_group_id attribute (instead of task.task_group.group_id), causing the discriminating group segment to be silently dropped from the key.

Changes:

• Fix _get_task_group_id in cosmos/operators/_watcher/xcom.py to access task.task_group.group_id directly (failing loudly on schema drift instead of silently).
• Add a regression integration test asserting two DbtTaskGroups yield distinct backup keys, and update/extend unit tests to use a spec'd mock so the bug can no longer be masked.
• Add an example DAG dev/dags/example_watcher_xcom_collision.py reproducing the original issue.

Reviewed changes

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

File	Description
cosmos/operators/_watcher/xcom.py	Reads task group id via the correct task.task_group.group_id path with direct attribute access.
tests/operators/_watcher/test_state.py	Updates unit test to mock the correct attribute path; adds a test ensuring malformed tasks raise.
tests/operators/test_watcher.py	Adds integration test verifying two watcher producers compute distinct backup keys.
dev/dags/example_watcher_xcom_collision.py	New example DAG reproducing the collision scenario end-to-end.

---

💡 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.03%. Comparing base (6cba83f) to head (6f4cfc3).
⚠️ Report is 2 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2683      +/-   ##
==========================================
+ Coverage   95.74%   98.03%   +2.29%     
==========================================
  Files         105      105              
  Lines        7864     7846      -18     
==========================================
+ Hits         7529     7692     +163     
+ Misses        335      154     -181     

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

Really clean fix — the root-cause analysis (silent getattr masking a non-existent attribute) and the three-layer regression strategy (real-attribute fix, MagicMock(spec=…) hardening, cross-group integration test) are excellent. Thanks for the thorough writeup!

One tiny doc nit: the PR description references example_watcher_xcom_collision.py as new, but it was removed in a follow-up commit and isn't in the final diff. Mind dropping the reference from the body so future readers don't go looking for it?

One cosmetic suggestion inline — nothing blocking.

---

🤖 Generated with Claude Code on behalf of @pankajastro

[pankajkoti]

Have a couple of questions in-line, but the fix for correcting the non-existent task_group_id -> task.task_group.group_id is a must, I agree.

[Copilot]

Pull request overview

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

[Copilot]

Pull request overview

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

[tatiana]

🚀 Released in Cosmos 1.14.2 (PyPI).