From 64235273b76e97dd29fd9dd7a47e537fd549a76b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joni=20P=C3=B6ll=C3=A4nen?= Date: Fri, 11 Sep 2026 17:48:28 +0300 Subject: [PATCH 1/4] Add paused_by --- task_manager/task_manager/active_tasks.py | 4 +-- task_manager/task_manager/task_client.py | 9 +++-- task_manager/task_manager/task_details.py | 1 + .../task_manager/task_manager_node.py | 3 +- .../tasks/composite_resolution.py | 3 +- task_manager/task_manager/tasks/mission.py | 8 ++--- .../task_manager/tasks/system_tasks.py | 9 ++--- .../task_manager_test_utils.py | 13 +++++-- .../integration_tests/test_system_tasks.py | 34 +++++++++++++++++++ 9 files changed, 66 insertions(+), 18 deletions(-) diff --git a/task_manager/task_manager/active_tasks.py b/task_manager/task_manager/active_tasks.py index d84a478..933b332 100644 --- a/task_manager/task_manager/active_tasks.py +++ b/task_manager/task_manager/active_tasks.py @@ -156,7 +156,7 @@ 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 @@ -166,7 +166,7 @@ def pause_task(self, task_id: str, publish: bool = True) -> None: :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 diff --git a/task_manager/task_manager/task_client.py b/task_manager/task_manager/task_client.py index 43e282e..d54bcb2 100644 --- a/task_manager/task_manager/task_client.py +++ b/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 diff --git a/task_manager/task_manager/task_details.py b/task_manager/task_manager/task_details.py index d567ad5..c31509c 100644 --- a/task_manager/task_manager/task_details.py +++ b/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) diff --git a/task_manager/task_manager/task_manager_node.py b/task_manager/task_manager/task_manager_node.py index 65615b6..f4e6d0f 100755 --- a/task_manager/task_manager/task_manager_node.py +++ b/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) diff --git a/task_manager/task_manager/tasks/composite_resolution.py b/task_manager/task_manager/tasks/composite_resolution.py index 62c1dfd..1ed8c81 100644 --- a/task_manager/task_manager/tasks/composite_resolution.py +++ b/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'. @@ -208,7 +209,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) diff --git a/task_manager/task_manager/tasks/mission.py b/task_manager/task_manager/tasks/mission.py index 4f60601..1b174fd 100644 --- a/task_manager/task_manager/tasks/mission.py +++ b/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 diff --git a/task_manager/task_manager/tasks/system_tasks.py b/task_manager/task_manager/tasks/system_tasks.py index 3736ef2..a936186 100644 --- a/task_manager/task_manager/tasks/system_tasks.py +++ b/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 diff --git a/task_manager/test/integration_tests/task_manager_test_utils.py b/task_manager/test/integration_tests/task_manager_test_utils.py index f169c1a..ed3b837 100644 --- a/task_manager/test/integration_tests/task_manager_test_utils.py +++ b/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,9 @@ 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]: + 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 +190,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) diff --git a/task_manager/test/integration_tests/test_system_tasks.py b/task_manager/test/integration_tests/test_system_tasks.py index 312723a..24593dd 100644 --- a/task_manager/test/integration_tests/test_system_tasks.py +++ b/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"): From c0971db6109e976e9dcfb0b9dcf2bba0d3a7e464 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joni=20P=C3=B6ll=C3=A4nen?= Date: Mon, 14 Sep 2026 15:16:34 +0300 Subject: [PATCH 2/4] Add docstring and messages which had not got pushed previously --- task_manager/task_manager/tasks/composite_resolution.py | 1 + task_manager_msgs/msg/ActiveTask.msg | 1 + task_manager_msgs/srv/PauseTasks.srv | 1 + 3 files changed, 3 insertions(+) diff --git a/task_manager/task_manager/tasks/composite_resolution.py b/task_manager/task_manager/tasks/composite_resolution.py index 1ed8c81..1c1a108 100644 --- a/task_manager/task_manager/tasks/composite_resolution.py +++ b/task_manager/task_manager/tasks/composite_resolution.py @@ -196,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. """ diff --git a/task_manager_msgs/msg/ActiveTask.msg b/task_manager_msgs/msg/ActiveTask.msg index 5bbc5aa..246bc2a 100644 --- a/task_manager_msgs/msg/ActiveTask.msg +++ b/task_manager_msgs/msg/ActiveTask.msg @@ -2,3 +2,4 @@ string task_id string task_name string task_status string source +string paused_by diff --git a/task_manager_msgs/srv/PauseTasks.srv b/task_manager_msgs/srv/PauseTasks.srv index 99b06ec..547e38f 100644 --- a/task_manager_msgs/srv/PauseTasks.srv +++ b/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 From 6b6adf91fdea15ee02b3ddb593d30ae120961082 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joni=20P=C3=B6ll=C3=A4nen?= Date: Mon, 14 Sep 2026 15:31:32 +0300 Subject: [PATCH 3/4] Added one more missing docstring --- task_manager/task_manager/active_tasks.py | 1 + 1 file changed, 1 insertion(+) diff --git a/task_manager/task_manager/active_tasks.py b/task_manager/task_manager/active_tasks.py index 933b332..0c811cc 100644 --- a/task_manager/task_manager/active_tasks.py +++ b/task_manager/task_manager/active_tasks.py @@ -162,6 +162,7 @@ def pause_task(self, task_id: str, publish: bool = True, paused_by: Optional[str :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. """ From 737dd68650b3dd4214f712445768855a5699c9bf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joni=20P=C3=B6ll=C3=A4nen?= Date: Mon, 14 Sep 2026 15:56:40 +0300 Subject: [PATCH 4/4] Fixing one more missing docstring --- task_manager/test/integration_tests/task_manager_test_utils.py | 1 + 1 file changed, 1 insertion(+) diff --git a/task_manager/test/integration_tests/task_manager_test_utils.py b/task_manager/test/integration_tests/task_manager_test_utils.py index ed3b837..784eaff 100644 --- a/task_manager/test/integration_tests/task_manager_test_utils.py +++ b/task_manager/test/integration_tests/task_manager_test_utils.py @@ -144,6 +144,7 @@ def wait_for_task_status(self, task_id: str, status: str, timeout: float = 5.0) 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(