Enable TaskIns for anonymous nodes

src/py/flwr/server/driver/state.py

@@ -48,7 +48,9 @@ def store_task_ins(self, task_ins: TaskIns) -> Optional[UUID]:
         # Return the new task_id
         return task_id
 
-    def get_task_ins(self, node_id: int, limit: Optional[int]) -> List[TaskIns]:
+    def get_task_ins(
+        self, node_id: Optional[int], limit: Optional[int]
+    ) -> List[TaskIns]:
         """Get all TaskIns that have not been delivered yet."""
 
         if limit is not None and limit < 1:
@@ -57,8 +59,16 @@ def get_task_ins(self, node_id: int, limit: Optional[int]) -> List[TaskIns]:
         # Find TaskIns for node_id that were not delivered yet
         task_ins_list: List[TaskIns] = []
         for _, task_ins in self.task_ins_store.items():
+            # pylint: disable=too-many-boolean-expressions
             if (
-                task_ins.task.consumer.node_id == node_id
+                node_id is not None  # Not anonymous
+                and task_ins.task.consumer.anonymous is False
+                and task_ins.task.consumer.node_id == node_id
+                and task_ins.task.delivered_at == ""
+            ) or (
+                node_id is None  # Anonymous
+                and task_ins.task.consumer.anonymous is True
+                and task_ins.task.consumer.node_id == 0
                 and task_ins.task.delivered_at == ""
             ):
                 task_ins_list.append(task_ins)

src/py/flwr/server/driver/state_test.py

@@ -39,6 +39,62 @@ def test_get_task_ins_empty() -> None:
     assert not task_ins_list
 
 
+def test_get_task_ins_identity() -> None:
+    """Validate that a new state has no TaskIns."""
+
+    # Prepare
+    state = DriverState()
+    task_id = state.store_task_ins(
+        task_ins=TaskIns(
+            task_id="",
+            group_id="",
+            workload_id="",
+            task=Task(
+                producer=Node(node_id=0, anonymous=True),
+                consumer=Node(node_id=123, anonymous=False),
+            ),
+        )
+    )
+
+    # Execute
+    task_ins_list = state.get_task_ins(
+        node_id=123,
+        limit=10,
+    )
+
+    # Assert
+    assert len(task_ins_list) == 1
+    assert task_ins_list[0].task_id == str(task_id)
+
+
+def test_get_task_ins_anonymous() -> None:
+    """Validate that a new state has no TaskIns."""
+
+    # Prepare
+    state = DriverState()
+    task_id = state.store_task_ins(
+        task_ins=TaskIns(
+            task_id="",
+            group_id="",
+            workload_id="",
+            task=Task(
+                producer=Node(node_id=0, anonymous=True),
+                consumer=Node(node_id=0, anonymous=True),
+            ),
+        )
+    )
+
+    # Execute
+    task_ins_list = state.get_task_ins(
+        node_id=None,
+        limit=10,
+    )
+
+    # Assert
+    assert len(task_ins_list) == 1
+    assert task_ins_list[0].task_id == str(task_id)
+
+
 def test_get_task_res_empty() -> None:
     """Validate that a new state has no TaskRes."""