fix: limit swarm callback concurrency

[yedidyakfir]

Summary

• Serialize swarm done/error callback tasks per swarm item using a Hatchet concurrency key made from swarm_task_id and swarm_item_id
• Keep fill_running_tasks concurrency behavior unchanged
• Add unit coverage for the internal Hatchet task registration options
• Clean up pipeline warnings from Mageflow unit tests and PR workflows
• Update the changelog for 0.3.6

Testing

• uv run pytest tests/unit -W error from libs/mageflow - 257 passed
• uv run pytest tests/unit/clients from libs/mageflow - 50 passed
• uv run pytest tests/unit/workflows from libs/mageflow - 27 passed
• uv run black --check mageflow/clients/hatchet/workflow.py mageflow/testing/_adapter.py mageflow/testing/plugin.py from libs/mageflow
• uv run ruff check mageflow/clients/hatchet/workflow.py mageflow/testing/_adapter.py mageflow/testing/plugin.py pyproject.toml from libs/mageflow

Notes

Linear issue lookup could not be completed because the Linear app connection requires reauthentication (oauth_token_invalid_grant), so this PR uses the fallback branch name from the implementation plan.

[coderabbitai]

Important

Review skipped

Auto incremental reviews are disabled on this repository.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: bd608752-e0c6-4d3d-baea-10d19171eb2e

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

📝 Walkthrough

Walkthrough

Added a SWARM_ITEM_ID_PARAM_NAME constant and applied a ConcurrencyExpression combining swarm task id and swarm item id to the ON_SWARM_ITEM_DONE and ON_SWARM_ITEM_ERROR durable tasks, throttling duplicate callback runs. Added corresponding unit tests and a changelog entry.

Changes

Swarm callback concurrency

Layer / File(s)	Summary
Concurrency constant and configuration libs/mageflow/mageflow/swarm/consts.py, libs/mageflow/mageflow/clients/hatchet/mageflow.py	Adds SWARM_ITEM_ID_PARAM_NAME constant and applies a per-(swarm task id + swarm item id) ConcurrencyExpression with max_runs=1 and GROUP_ROUND_ROBIN to the ON_SWARM_ITEM_DONE/ON_SWARM_ITEM_ERROR tasks, which previously had no concurrency limit.
Tests and changelog libs/mageflow/tests/unit/clients/test_hatchet_mageflow_tasks.py, CHANGELOG.md	New unit tests verify the callback concurrency gate and that SWARM_FILL_TASK's existing two-entry concurrency configuration is preserved; changelog documents the fix.

Estimated code review effort: 2 (Simple) | ~10 minutes

Sequence Diagram(s)

sequenceDiagram
  participant Hatchet
  participant HatchetMageflow
  participant OnSwarmItemDone

  Hatchet->>HatchetMageflow: init_mageflow_hatchet_tasks()
  HatchetMageflow->>OnSwarmItemDone: register task with ConcurrencyExpression(swarm_task_id, swarm_item_id)
  Hatchet-->>OnSwarmItemDone: duplicate done/error trigger
  OnSwarmItemDone-->>OnSwarmItemDone: run serialized via GROUP_ROUND_ROBIN (max_runs=1)

Loading

Related issues: None specified.

Suggested labels: bug, mageflow

Suggested reviewers: None specified.

Poem:
A rabbit hopped through swarm and queue,
Two callbacks raced, but only one broke through.
Now item and task IDs form a gate,
Round-robin style, no duplicate fate. 🐇

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title accurately summarizes the main change: limiting swarm callback concurrency to prevent duplicate callback runs.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch codex/hatchet-swarm-callback-lock

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[coderabbitai]

🧹 Nitpick comments (1)

libs/mageflow/tests/unit/clients/test_hatchet_mageflow_tasks.py (1)

21-27: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Duplicated fake_durable_task test double.

Same closure is redefined in both tests. Consider extracting it into a shared fixture/helper to avoid drift between the two copies.

♻️ Example fixture extraction

+@pytest.fixture
+def registered_tasks_recorder(hatchet_mock, monkeypatch):
+    registered_tasks = {}
+
+    def fake_durable_task(**kwargs):
+        registered_tasks[kwargs["name"]] = kwargs
+
+        def decorator(func):
+            return func
+
+        return decorator
+
+    monkeypatch.setattr(hatchet_mock, "durable_task", fake_durable_task)
+    return registered_tasks

Also applies to: 69-75

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@libs/mageflow/tests/unit/clients/test_hatchet_mageflow_tasks.py` around lines
21 - 27, The test double fake_durable_task is duplicated across the two tests,
so extract it into a shared helper or pytest fixture and reuse it from both test
cases. Keep the behavior centralized around the registered_tasks capture and
decorator closure so future changes only need to be made in one place, and
update the tests to reference the shared helper instead of redefining
fake_durable_task locally.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Nitpick comments:
In `@libs/mageflow/tests/unit/clients/test_hatchet_mageflow_tasks.py`:
- Around line 21-27: The test double fake_durable_task is duplicated across the
two tests, so extract it into a shared helper or pytest fixture and reuse it
from both test cases. Keep the behavior centralized around the registered_tasks
capture and decorator closure so future changes only need to be made in one
place, and update the tests to reference the shared helper instead of redefining
fake_durable_task locally.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: 380a8f23-84d0-4a74-91e7-a99028fd5743

📥 Commits

Reviewing files that changed from the base of the PR and between 6b6107a and 55eff4e.

📒 Files selected for processing (4)

• CHANGELOG.md
• libs/mageflow/mageflow/clients/hatchet/mageflow.py
• libs/mageflow/mageflow/swarm/consts.py
• libs/mageflow/tests/unit/clients/test_hatchet_mageflow_tasks.py