Pausing ParallelTask with a service

task_manager/task_manager/tasks/parallel_task_executor.py

@@ -173,7 +173,6 @@ def _wait_actions_done(self, goal_handle: ServerGoalHandle, subtasks: List[Paral
         :param goal_handle: Handle of the parallel goal
         :param subtasks: Reference to the list of tasks to wait for.
         """
-        goal_id = bytes(goal_handle.goal_id.uuid)
         while rclpy.ok():
             if goal_handle.is_cancel_requested:
                 self._logger.info("Cancel requested for parallel task.")
@@ -186,17 +185,10 @@ def _wait_actions_done(self, goal_handle: ServerGoalHandle, subtasks: List[Paral
             if self._latest_goal_handle != goal_handle:
                 raise PreemptedException("Parallel task was preempted")
 
-            # While paused, a subtask reaching completion on its own (e.g. a service-backed one that couldn't
-            # actually be paused and simply ran to completion) must not tear down the rest of the group.
-            if not self.is_paused(goal_id):
-                for action in subtasks:
-                    # `has_finished()` is the real completion signal - a paused subtask is neither
-                    # active nor finished, and must not tear the group down.
-                    if action.has_finished():
-                        self._logger.info(
-                            f"Action '{action.name}' finished. The whole parallel action will be cancelled"
-                        )
-                        return
+            for action in subtasks:
+                if action.has_finished():
+                    self._logger.info(f"Action '{action.name}' finished. The whole parallel action will be cancelled")
+                    return
 
             time.sleep(0.1)
 

task_manager/test/integration_tests/test_parallel_task_executor.py

@@ -380,6 +380,29 @@ def test_pause_and_resume_parallel_task_by_its_own_id(self):
         self.execute_cancel_task([parallel_id])
         goal_handle.get_result()
 
+    def test_pause_tears_down_the_sibling_once_the_service_subtask_finishes(self):
+        """Test that pausing a parallel group tears down the sibling once the service subtask finishes."""
+        goal = PerformInParallel.Goal(
+            subtasks=[
+                SubtaskGoal(task_id="fib", task_name="fibonacci", task_data='{"order": 10}'),
+                SubtaskGoal(task_id="add", task_name="add_two_ints", task_data='{"a": 0, "b": 1}'),
+            ]
+        )
+        goal_handle = self.run_parallel_tasks(goal)
+        self.wait_for_task_start("fib")
+        self.wait_for_task_start("add")
+        parallel_id = self._active_task_ids_by_name()["system/perform_in_parallel"]
+
+        pause_response = self.execute_pause_task([parallel_id])
+        self.assertEqual(pause_response.result.task_status, TaskStatus.DONE)
+
+        self.wait_for_task_status("add", TaskStatus.DONE)
+        self.wait_for_task_status("fib", TaskStatus.CANCELED)
+        self.wait_for_task_status(parallel_id, TaskStatus.DONE)
+
+        result = goal_handle.get_result()
+        self.assertEqual(result.result.task_status, TaskStatus.DONE)
+
     @staticmethod
     def parse_task_results(task_results: str) -> Dict[str, Dict[str, Any]]:
         """Parse the task results from a JSON string to a dictionary with task_id as key and the result as value."""

task_manager/test/test_parallel_task_executor_unit.py

@@ -157,11 +157,10 @@ def test_request_pause_resume_is_paused_state_transitions(parallel_task_executor
     assert parallel_task_executor.is_paused(goal_id) is False
 
 
-def test_wait_actions_done_ignores_a_finished_member_while_paused(parallel_task_executor: ParallelTaskExecutor) -> None:
-    """ParallelTask should not tear down if subtask finishes when pausing.
-
-    ParallelTask should restart everything after pause.
-    """
+def test_wait_actions_done_completes_when_the_only_subtask_finished_while_paused(
+    parallel_task_executor: ParallelTaskExecutor,
+) -> None:
+    """Service tasks cannot be paused and should run to completion immediately."""
     goal_id = b"\x03" * 16
     goal_handle = MagicMock()
     goal_handle.is_cancel_requested = False
@@ -178,17 +177,41 @@ def test_wait_actions_done_ignores_a_finished_member_while_paused(parallel_task_
     thread = threading.Thread(target=parallel_task_executor._wait_actions_done, args=(goal_handle, [task]))
     thread.start()
     try:
-        time.sleep(0.3)
-        assert thread.is_alive()  # still waiting - the finished member must not have torn the group down
-
-        parallel_task_executor.request_resume(goal_id)
         thread.join(timeout=2)
         assert not thread.is_alive()
     finally:
         goal_handle.is_active = False
         thread.join(timeout=2)
 
 
+def test_wait_actions_done_tears_down_when_a_member_finishes(
+    parallel_task_executor: ParallelTaskExecutor,
+) -> None:
+    """Test that _wait_actions_done tears down immediately when a member finishes."""
+    goal_id = b"\x06" * 16
+    goal_handle = MagicMock()
+    goal_handle.is_cancel_requested = False
+    goal_handle.is_active = True
+    goal_handle.goal_id.uuid = list(goal_id)
+
+    finished_task = make_parallel_task(TaskStatus.DONE, task_id="finished")
+    finished_task._task_client.goal_done = True
+
+    paused_task = make_parallel_task(TaskStatus.PAUSED, task_id="paused")
+    paused_task._task_client.goal_done = False
+
+    parallel_task_executor._latest_goal_handle = goal_handle
+    _register(parallel_task_executor, goal_id)
+    parallel_task_executor.request_pause(goal_id)
+
+    thread = threading.Thread(
+        target=parallel_task_executor._wait_actions_done, args=(goal_handle, [finished_task, paused_task])
+    )
+    thread.start()
+    thread.join(timeout=2)
+    assert not thread.is_alive()
+
+
 def test_wait_actions_done_cancel_still_ends_it_while_paused(parallel_task_executor: ParallelTaskExecutor) -> None:
     """A genuine cancel of the group's own goal must end _wait_actions_done immediately, regardless of whether the group
     is currently paused."""