diff --git a/CHANGELOG.md b/CHANGELOG.md index 6d0241f8..3cd74249 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ - **Keychain Access Error Handling**: The app now distinguishes between missing credentials (first launch) and macOS blocking keychain access (e.g. after app rename/update). A dedicated error screen guides the user to re-enter credentials instead of silently dropping into onboarding. - **Swarm Callback Concurrency**: Duplicate swarm done/error callbacks are now serialized per swarm item using the swarm and item IDs. +- **Empty Swarm Task Extraction**: Swarm filling now returns before querying Rapyer when there are no task IDs, preventing an empty variadic lookup from scanning the Redis database. ### 🔄 Changed diff --git a/libs/mageflow/mageflow/swarm/workflows.py b/libs/mageflow/mageflow/swarm/workflows.py index 97e2806d..3180bafb 100644 --- a/libs/mageflow/mageflow/swarm/workflows.py +++ b/libs/mageflow/mageflow/swarm/workflows.py @@ -105,25 +105,26 @@ async def fill_running_tasks( publish_state.task_ids.extend(task_ids_to_run) swarm.tasks_left_to_run.remove_range(0, num_of_task_to_run) - if task_ids_to_run: - tasks = await rapyer.afind(*task_ids_to_run) - tasks = cast(list[Signature], tasks) - - # Update the kwargs locally, so swarm kwargs wont be duplicated on redis but still sent to task - swarm_kwargs = swarm.kwargs.copy() - swarm_msg = swarm_kwargs.pop(SWARM_MESSAGE_PARAM_NAME, None) - for task in tasks: - task.kwargs.update(**swarm_kwargs) - - await swarm.ClientAdapter.acall_signatures( - tasks, - swarm_msg, - set_return_field=swarm.config.send_swarm_message_to_return_field, - **pub_kwargs, - ) - - async with publish_state.apipeline(): - publish_state.task_ids.clear() - swarm.current_running_tasks += num_of_task_to_run - return tasks - return [] + if not task_ids_to_run: + return [] + + tasks = await rapyer.afind(*task_ids_to_run) + tasks = cast(list[Signature], tasks) + + # Update the kwargs locally, so swarm kwargs wont be duplicated on redis but still sent to task + swarm_kwargs = swarm.kwargs.copy() + swarm_msg = swarm_kwargs.pop(SWARM_MESSAGE_PARAM_NAME, None) + for task in tasks: + task.kwargs.update(**swarm_kwargs) + + await swarm.ClientAdapter.acall_signatures( + tasks, + swarm_msg, + set_return_field=swarm.config.send_swarm_message_to_return_field, + **pub_kwargs, + ) + + async with publish_state.apipeline(): + publish_state.task_ids.clear() + swarm.current_running_tasks += num_of_task_to_run + return tasks diff --git a/libs/mageflow/tests/unit/workflows/test_fill_running_tasks.py b/libs/mageflow/tests/unit/workflows/test_fill_running_tasks.py index b47cc877..a4efd807 100644 --- a/libs/mageflow/tests/unit/workflows/test_fill_running_tasks.py +++ b/libs/mageflow/tests/unit/workflows/test_fill_running_tasks.py @@ -1,3 +1,5 @@ +from unittest.mock import patch + import pytest import thirdmagic from hatchet_sdk.clients.admin import TriggerWorkflowOptions @@ -8,6 +10,20 @@ from tests.integration.hatchet.models import ContextMessage +@pytest.mark.asyncio +async def test_fill_running_tasks_does_not_extract_when_task_ids_are_empty( + empty_swarm, mock_adapter +): + # Act + with patch("rapyer.afind", side_effect=AssertionError) as mock_afind: + tasks = await fill_running_tasks(empty_swarm) + + # Assert + assert tasks == [] + mock_afind.assert_not_called() + mock_adapter.acall_signatures.assert_not_awaited() + + @pytest.mark.asyncio @pytest.mark.parametrize( [