Guard empty swarm task extraction

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
 

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

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(
     [