Add paused_by information to ActiveTasks

task_manager/task_manager/active_tasks.py

@@ -156,17 +156,18 @@ def cancel_task(self, task_id: str) -> None:
         task_client = self.get_task_client(task_id)
         task_client.cancel_task()
 
-    def pause_task(self, task_id: str, publish: bool = True) -> None:
+    def pause_task(self, task_id: str, publish: bool = True, paused_by: Optional[str] = None) -> None:
         """Pauses the active task based on task ID.
 
         :param task_id: ID of the task to pause
         :param publish: If False, skip republishing after pausing - use when the caller has more related changes
             to make (e.g. mirroring the status onto an owning Mission) and will publish once itself afterward.
+        :param paused_by: the identifier of the entity requesting the pause or reason for pausing, if applicable.
         :raises KeyError: if a task with the given id was not found.
         :raises PauseTaskFailedError: if pausing of the task fails.
         """
         task_client = self.get_task_client(task_id)
-        task_client.pause_task()
+        task_client.pause_task(paused_by=paused_by)
         if publish:
             self.publish_active_tasks()  # Republish so the PAUSED status is visible on the active tasks topic
 

task_manager/task_manager/task_client.py

@@ -82,7 +82,7 @@ def request_canceling(self) -> None:
         """Request canceling the task asynchronously."""
 
     @abstractmethod
-    def pause_task(self) -> None:
+    def pause_task(self, paused_by: Optional[str] = None) -> None:
         """Pause the task synchronously."""
 
     @abstractmethod
@@ -330,7 +330,7 @@ def _handle_cancel_response(self, response: CancelGoal.Response) -> None:
                 )
                 raise CancelTaskFailedError("Couldn't cancel the task!")
 
-    def pause_task(self) -> None:
+    def pause_task(self, paused_by: Optional[str] = None) -> None:
         """Pauses the task by cancelling the underlying action goal, keeping the original goal message stored so that
         resume_task() can restart it later.
 
@@ -370,6 +370,7 @@ def pause_task(self) -> None:
         self._pausing = False
         self._paused = True
         self.task_details.status = TaskStatus.PAUSED
+        self.task_details.paused_by = paused_by if paused_by else ""
 
     def resume_task(self) -> None:
         """Resumes a paused task by re-sending the original goal message as a brand-new action goal.
@@ -386,6 +387,7 @@ def resume_task(self) -> None:
         self._result_future = None
         try:
             self.start_task_async(self._last_goal_message)
+            self.task_details.paused_by = ""
         except TaskStartError as e:
             # start_task_async() already set status to ERROR, but goal_done is only ever set via the done-callback
             # chain, which nothing else will trigger for a start failure - without this, the task would stay
@@ -604,7 +606,7 @@ def request_canceling(self) -> None:
             f"Cancel call to service {self._task_specs.topic} is not supported. Ignoring the cancel request."
         )
 
-    def pause_task(self) -> None:
+    def pause_task(self, paused_by: Optional[str] = None) -> None:
         """Services cannot be cancelled mid-flight, so instead of failing immediately, this waits out cancel_timeout for
         the ongoing call to finish naturally.
 
@@ -614,6 +616,7 @@ def pause_task(self) -> None:
 
         :raises PauseTaskFailedError: If the service call is still running after cancel_timeout.
         """
+        del paused_by  # To satisfy linters (argument is not used here)
         if self._goal_done.is_set():
             return
 

task_manager/task_manager/task_details.py

@@ -35,6 +35,7 @@ class TaskDetails:
     task_id: str
     source: str
     status: TaskStatus
+    paused_by: str = field(default="")
 
     # Other public fields (not constructor args)
     result: Optional[Any] = field(default=None, init=False)

task_manager/task_manager/task_manager_node.py

@@ -193,8 +193,8 @@ def setup_system_tasks(self):  # pylint: disable=too-many-locals
             resume_topic = "_" + resume_topic
 
         stop_service = StopTasksService(self, topic=stop_topic, active_tasks=self.active_tasks)
-        mission = Mission(self, action_name=mission_topic, execute_task_cb=self.execute_task)
         wait = WaitTask(self, topic=wait_topic)
+        mission = Mission(self, action_name=mission_topic, execute_task_cb=self.execute_task)
         parallel = ParallelTaskExecutor(
             self,
             topic=parallel_topic,
@@ -346,6 +346,7 @@ def _active_tasks_changed_cb(self, active_tasks: Dict[str, TaskClient]) -> None:
             task_msg.task_name = task_client.task_specs.task_name
             task_msg.task_status = str(task_client.task_details.status)
             task_msg.source = task_client.task_details.source
+            task_msg.paused_by = task_client.task_details.paused_by
             task_messages.append(task_msg)
         msg = ActiveTaskArray(active_tasks=task_messages)
         self._active_tasks_pub.publish(msg)

task_manager/task_manager/tasks/composite_resolution.py

@@ -178,6 +178,7 @@ def pause_or_resume_group(  # pylint: disable=too-many-arguments, too-many-posit
     start_statuses: Tuple[TaskStatus, ...],
     callback: Callable[[str], bool],
     pause: bool,
+    paused_by: Optional[str] = None,
 ) -> bool:
     """Run function 'callback' on every task which is linked to the task with given 'task_id'.
 
@@ -195,6 +196,7 @@ def pause_or_resume_group(  # pylint: disable=too-many-arguments, too-many-posit
         if the transition succeeded, False if it failed.
     :param pause: True to arm every composite's own paused flag encountered while resolving down to the leaves,
         False to disarm them - see resolve_down's `paused_flag` parameter.
+    :param paused_by: the identifier of the entity requesting the pause or reason for pausing, if applicable.
     :raises KeyError: if task_id (or its resolved starting point) is not an active task.
     :return: True if everything succeeded.
     """
@@ -208,7 +210,7 @@ def pause_or_resume_group(  # pylint: disable=too-many-arguments, too-many-posit
             continue  # Finished on its own - not a failure (matches this function's documented contract).
         if member_status not in start_statuses:
             continue  # Already in the target state, or finished on its own - nothing to do, not a failure.
-        if not callback(member_id):
+        if not callback(member_id, paused_by):
             success = False
 
     _sync_composite_statuses(active_tasks, composites)

task_manager/task_manager/tasks/mission.py

@@ -50,7 +50,7 @@ def __init__(
         super().__init__()
         self.execute_task_cb = execute_task_cb
         # ROS action goal_id (as bytes) of each running mission -> task_id of its currently running subtask.
-        self._goal_id_to_subtask_id: Dict[bytes, str] = {}
+        self._subtasks_per_mission: Dict[bytes, str] = {}
         ActionServer(
             node=node,
             action_type=MissionAction,
@@ -63,7 +63,7 @@ def __init__(
     def get_active_children(self, goal_id: bytes) -> List[str]:
         """Returns the mission's active children i.e. its single currently-running subtask or an empty list if that
         mission subtask isn't running (or isn't known)."""
-        current = self._goal_id_to_subtask_id.get(goal_id)
+        current = self._subtasks_per_mission.get(goal_id)
         return [current] if current is not None else []
 
     def _wait_until_resumed(self, goal_id: bytes, goal_handle: ServerGoalHandle) -> bool:
@@ -92,7 +92,7 @@ def execute_cb(self, goal_handle):
                     goal_handle.canceled()
                     return result
 
-                self._goal_id_to_subtask_id[goal_id] = subtask.task_id
+                self._subtasks_per_mission[goal_id] = subtask.task_id
                 goal = ExecuteTask.Goal(
                     task_id=subtask.task_id, task_name=subtask.task_name, task_data=subtask.task_data, source="Mission"
                 )
@@ -120,7 +120,7 @@ def execute_cb(self, goal_handle):
             goal_handle.succeed()
             return result
         finally:
-            self._goal_id_to_subtask_id.pop(goal_id, None)
+            self._subtasks_per_mission.pop(goal_id, None)
             self._stop_pause_tracking(goal_id)
 
     @staticmethod

task_manager/task_manager/tasks/system_tasks.py

@@ -15,7 +15,7 @@
 #  ------------------------------------------------------------------
 import time
 from abc import ABC, abstractmethod
-from typing import Dict, List
+from typing import Dict, List, Optional
 
 # ROS
 import rclpy
@@ -179,9 +179,9 @@ def __init__(
             PauseTasks, self._topic, self.service_cb, callback_group=MutuallyExclusiveCallbackGroup()
         )
 
-    def _try_pause(self, task_id: str) -> bool:
+    def _try_pause(self, task_id: str, _paused_by: Optional[str] = None) -> bool:
         try:
-            self._active_tasks.pause_task(task_id, publish=False)
+            self._active_tasks.pause_task(task_id, publish=False, paused_by=_paused_by)
             return True
         except PauseTaskFailedError as e:
             self._node.get_logger().error(f"Failed to pause task with ID {task_id}: {e}")
@@ -200,6 +200,7 @@ def service_cb(self, request: PauseTasks.Request, response: PauseTasks.Response)
                     (TaskStatus.RECEIVED, TaskStatus.IN_PROGRESS),
                     self._try_pause,
                     pause=True,
+                    paused_by=request.paused_by,
                 )
             except KeyError:
                 self._node.get_logger().error(f"Tried to pause a task with ID {task_id}, but the task is not active.")
@@ -253,7 +254,7 @@ def __init__(
             ResumeTasks, self._topic, self.service_cb, callback_group=MutuallyExclusiveCallbackGroup()
         )
 
-    def _try_resume(self, task_id: str) -> bool:
+    def _try_resume(self, task_id: str, _paused_by: Optional[str] = None) -> bool:
         try:
             self._active_tasks.resume_task(task_id, publish=False)
             return True

task_manager/test/integration_tests/task_manager_test_utils.py

@@ -20,7 +20,7 @@
 import unittest
 import uuid
 from asyncio import Future
-from typing import List, Optional
+from typing import Dict, List, Optional
 
 # ROS
 import rclpy
@@ -41,7 +41,7 @@
 
 # Task Manager messages
 from task_manager_msgs.action import ExecuteTask, PerformInParallel, Wait
-from task_manager_msgs.msg import ActiveTaskArray
+from task_manager_msgs.msg import ActiveTask, ActiveTaskArray
 from task_manager_msgs.srv import CancelTasks, PauseTasks, ResumeTasks, StopTasks
 
 # Task Manager
@@ -106,12 +106,14 @@ def setUp(self) -> None:
         )
         self._tasks_started = []
         self._task_statuses = {}
+        self._current_active_tasks = {}
 
     def _active_tasks_cb(self, active_tasks_msg) -> None:
         for active_task in active_tasks_msg.active_tasks:
             if active_task.task_id not in self._tasks_started:
                 self._tasks_started.append(active_task.task_id)
             self._task_statuses[active_task.task_id] = active_task.task_status
+            self._current_active_tasks[active_task.task_id] = active_task
 
     def tearDown(self) -> None:
         self.task_manager_node.destroy_node()
@@ -141,6 +143,10 @@ def wait_for_task_status(self, task_id: str, status: str, timeout: float = 5.0)
             assert time.time() - start < timeout
             time.sleep(0.01)
 
+    def get_active_tasks(self) -> Dict[str, ActiveTask]:
+        """Returns the current active tasks as a dictionary mapping task IDs to ActiveTask instances."""
+        return self._current_active_tasks
+
     def start_fibonacci_action_task(
         self, task_name: str = "fibonacci", run_time_secs: int = 3, task_id: Optional[str] = None
     ) -> ClientGoalHandle:
@@ -185,10 +191,12 @@ def execute_cancel_task(self, task_ids: List[str]) -> ExecuteTask.Result:
         goal_handle = self._start_task(goal=goal)
         return goal_handle.get_result()
 
-    def execute_pause_task(self, task_ids: List[str]) -> ExecuteTask.Result:
+    def execute_pause_task(self, task_ids: List[str], paused_by: Optional[str] = None) -> ExecuteTask.Result:
         """Calls system/pause_task with given task IDs."""
         pause_goal = PauseTasks.Request()
         pause_goal.paused_tasks = task_ids
+        if paused_by:
+            pause_goal.paused_by = paused_by
         task_data = json.dumps(extract_values(pause_goal))
         goal = ExecuteTask.Goal(task_name="system/pause_task", task_data=task_data, source="")
         goal_handle = self._start_task(goal=goal)

task_manager/test/integration_tests/test_system_tasks.py

@@ -198,6 +198,40 @@ def test_resume_task_while_another_blocking_task_is_running(self) -> None:
         # The resumed task must still finish normally
         self.assertEqual(goal_handle_1.get_result().result.task_status, TaskStatus.DONE)
 
+    def test_paused_by_handling_on_pause(self):
+        """When paused and if the paused_by information is provided, it should be set in the task details."""
+        _goal_handle = self.start_fibonacci_action_task("fibonacci_blocking", run_time_secs=1, task_id="111")
+        self.wait_for_task_start("111")
+
+        paused_by = "test_user"
+        pause_response = self.execute_pause_task(task_ids=["111"], paused_by=paused_by)
+        self.assertEqual(pause_response.result.task_status, TaskStatus.DONE)
+        self.wait_for_task_status("111", TaskStatus.PAUSED)
+
+        active_tasks = self.get_active_tasks()
+        self.assertIn("111", active_tasks)
+        self.assertEqual(active_tasks["111"].paused_by, paused_by)
+
+        # Cleanup
+        self.execute_cancel_task(task_ids=["111"])
+
+    def test_paused_by_handling_on_resume(self):
+        """On resume, the paused_by information should be cleared."""
+        _goal_handle = self.start_fibonacci_action_task("fibonacci_blocking", run_time_secs=1, task_id="111")
+        self.wait_for_task_start("111")
+
+        paused_by = "test_user"
+        self.execute_pause_task(task_ids=["111"], paused_by=paused_by)
+        self.wait_for_task_status("111", TaskStatus.PAUSED)
+
+        resume_response = self.execute_resume_task(task_ids=["111"])
+        self.assertEqual(resume_response.result.task_status, TaskStatus.DONE)
+        self.wait_for_task_status("111", TaskStatus.DONE)
+
+        active_tasks = self.get_active_tasks()
+        self.assertIn("111", active_tasks)
+        self.assertEqual(active_tasks["111"].paused_by, "")
+
     def test_stop_task(self) -> None:
         """Test cases for Stop system task."""
         with self.subTest("Task with 'cancel_on_stop' field is cancelled"):

task_manager_msgs/msg/ActiveTask.msg

@@ -2,3 +2,4 @@ string task_id
 string task_name
 string task_status
 string source
+string paused_by

task_manager_msgs/srv/PauseTasks.srv

@@ -1,4 +1,5 @@
 string[] paused_tasks               # Task IDs of the active tasks to be paused
+string paused_by                    # The identifier of the entity requesting the pause or reason for pausing
 ---
 bool success                        # Success if all the pauses pass
 string[] successful_pauses          # List of successfully paused tasks