Add runtime dependency inference from controller manager

foreman/config/scenario.yaml

@@ -5,17 +5,6 @@ hardware:
 lifecycle_nodes:
   - dummy_lifecycle_node
 
-controllers:
-  joint_state_broadcaster:
-    requires:
-      - [kassow, inactive]
-      - [FrankaHardwareInterface, inactive]
-      - [dummy_lifecycle_node, active]
-  kassow_joint_trajectory_controller:
-    requires: [kassow, active]
-  franka_joint_trajectory_controller:
-    requires: [FrankaHardwareInterface, active]
-
 goal_states:
 
   idle:

foreman/foreman/adapters/component_state_monitor.py

@@ -1,6 +1,9 @@
+import threading
 from typing import Dict, List
 
 from controller_manager_msgs.msg import ControllerManagerActivity
+from controller_manager_msgs.srv import ListControllers
+from controller_manager_msgs.srv import ListHardwareComponents
 from lifecycle_msgs.msg import TransitionEvent
 from lifecycle_msgs.srv import GetState
 from rclpy.event_handler import QoSSubscriptionMatchedInfo
@@ -14,6 +17,8 @@
 from foreman.engine import ForemanEngine
 from foreman.types import Component
 from foreman.types import ComponentType
+from foreman.types import ControllerDependencyRule
+from foreman.types import HardwareRequirement
 from foreman.types import LifecycleState
 
 
@@ -64,6 +69,24 @@ def __init__(
             f"{self._logger_prefix} Subscribed to /{controller_manager_name}/activity"
         )
 
+        # Service clients used to query controller_manager when building dependency rules.
+        self._client_list_controllers = self._node.create_client(
+            ListControllers,
+            f'/{controller_manager_name}/list_controllers',
+            callback_group=self._node.callback_group_subscriber
+        )
+        self._client_list_hardware_components = self._node.create_client(
+            ListHardwareComponents,
+            f'/{controller_manager_name}/list_hardware_components',
+            callback_group=self._node.callback_group_subscriber
+        )
+
+        # Latest controller_manager observations used to infer dependency rules.
+        # Updated asynchronously from controller_manager callbacks.
+        self._dependency_lock = threading.Lock()
+        self._latest_controllers = []
+        self._latest_hardware_components = []
+
         # --- Lifecycle node monitoring ---
         self._lc_node_get_state_clients: Dict[str, object] = {}
 
@@ -96,6 +119,88 @@ def __init__(
                 f"{self._logger_prefix} Monitoring lifecycle nodes: {lifecycle_nodes}"
             )
 
+    @staticmethod
+    def infer_dependency_rules(controller_states, hardware_components):
+        """Infer each controller's hardware dependencies from its required interfaces.
+
+        A command interface requires its owning hardware ACTIVE; a state interface
+        requires it only INACTIVE. If a controller needs both from one hardware, ACTIVE wins.
+        """
+        owner_of_interface = {}
+        # Build a mapping of interface names to their owning hardware components
+        for hardware in hardware_components:
+            for interface in list(hardware.command_interfaces) + list(hardware.state_interfaces):
+                owner_of_interface[interface.name] = hardware.name
+
+        dependency_rules = []
+        for controller in controller_states:
+            # Map required interfaces to hardware components and their required lifecycle states
+            required_hardware_state = {}
+            for interface in controller.required_command_interfaces:
+                hardware_name = owner_of_interface.get(interface)
+                if hardware_name:
+                    required_hardware_state[hardware_name] = LifecycleState.ACTIVE
+
+            for interface in controller.required_state_interfaces:
+                hardware_name = owner_of_interface.get(interface)
+                if hardware_name and hardware_name not in required_hardware_state:
+                    required_hardware_state[hardware_name] = LifecycleState.INACTIVE
+
+            dependency_rules.append(ControllerDependencyRule(
+                controller_name=controller.name,
+                required_hardware=[
+                    HardwareRequirement(name=name, state=state)
+                    for name, state in required_hardware_state.items()
+                ],
+            ))
+        return dependency_rules
+
+    def get_dependency_rules(self):
+        """
+        Infer dependency rules from the latest controller_manager observations.
+
+        The observations are refreshed asynchronously, so this method never blocks.
+        """
+        with self._dependency_lock:
+            return self.infer_dependency_rules(
+                self._latest_controllers, self._latest_hardware_components)
+
+    def _refresh_dependency_rules(self):
+        """Refresh dependency observations asynchronously.
+
+        Query hardware first, then controllers, before rebuilding dependencies.
+        """
+        if not (self._client_list_hardware_components.service_is_ready()
+                and self._client_list_controllers.service_is_ready()):
+            return
+        future = self._client_list_hardware_components.call_async(
+            ListHardwareComponents.Request())
+        future.add_done_callback(self._on_hardware_components_response)
+
+    def _on_hardware_components_response(self, future):
+        """Store the latest hardware observations and request controllers."""
+        try:
+            hardware = future.result().component
+        except Exception as e:
+            self._node.get_logger().warning(
+                f"{self._logger_prefix} Failed to list hardware components: {e}")
+            return
+        with self._dependency_lock:
+            self._latest_hardware_components = hardware
+        ctrl_future = self._client_list_controllers.call_async(ListControllers.Request())
+        ctrl_future.add_done_callback(self._on_controllers_response)
+
+    def _on_controllers_response(self, future):
+        """Store the latest controller observations."""
+        try:
+            controllers = future.result().controller
+        except Exception as e:
+            self._node.get_logger().warning(
+                f"{self._logger_prefix} Failed to list controllers: {e}")
+            return
+        with self._dependency_lock:
+            self._latest_controllers = controllers
+
     # TODO: use matched event for this topic as well? That way we know if controller manager dies.
     # Minor. Currently we catch unexpected transitions in the engine (all components go to finalized)
     def _activity_callback(self, msg: ControllerManagerActivity):
@@ -124,6 +229,8 @@ def _activity_callback(self, msg: ControllerManagerActivity):
 
         self._cm_components = components
         self._push_merged_state()
+        # Refresh dependency observations after receiving a new activity update.
+        self._refresh_dependency_rules()
 
     def _on_lifecycle_publisher_matched(self, name: str, info: QoSSubscriptionMatchedInfo):
         """DDS matched event: lifecycle node's transition_event publisher appeared or disappeared."""

foreman/foreman/engine.py

@@ -5,6 +5,8 @@
 from foreman.planner import Planner
 from foreman.types import Component
 from foreman.types import ComponentType
+from foreman.types import ControllerDependencyRule
+from foreman.types import DependencyProvider
 from foreman.types import ErrorSnapshot
 from foreman.types import ForemanError
 from foreman.types import ForemanErrorCategory
@@ -91,6 +93,10 @@ def abort_goal(self, error: ForemanError):
             self._last_issued_command = None
             self._locked_abort_transition()
 
+    def set_dependency_provider(self, dependency_provider: DependencyProvider):
+        """Set the source of dependency rules for the planner."""
+        self._planner.set_dependency_provider(dependency_provider)
+
     def get_next_transition(self) -> Optional[SystemTransitionCommand]:
         """Calculate the next step toward the goal."""
         if not self._current_goal:
@@ -251,8 +257,9 @@ def _locked_check_unsatisfiable_dependencies(self, goal: SystemGoal) -> List[str
             goal_infrastructure_states[comp.name] = comp.lifecycle_state
 
         errors = []
+        current_dependency_rules = self._planner.get_current_rules()
         for ctrl_goal in goal.controller_goals:
-            rule = self._planner.rules.get(ctrl_goal.name)
+            rule = current_dependency_rules.get(ctrl_goal.name)
             if not rule:
                 continue
 

foreman/foreman/node.py

@@ -47,6 +47,8 @@ def __init__(self):
             controller_manager_name=controller_manager_name,
             lifecycle_nodes=self.foreman_config.lifecycle_nodes
         )
+        # Planner queries dependency rules from the monitor when planning.
+        self.foreman_engine.set_dependency_provider(self.component_state_monitor)
         self.controller_manager_service_caller = adapters.ControllerManagerServiceCaller(
             node=self,
             controller_manager_name=controller_manager_name

foreman/foreman/parser.py

@@ -8,7 +8,6 @@
 from foreman.types import Component
 from foreman.types import ComponentType
 from foreman.types import ControllerDependencyRule
-from foreman.types import HardwareRequirement
 from foreman.types import LifecycleState
 from foreman.types import SystemGoal
 
@@ -44,49 +43,6 @@ def parse_state_string(state_str: str) -> LifecycleState:
     return state_mapping[normalized]
 
 
-def parse_requires(
-    requires: List[str], hardware: List[str], lifecycle_nodes: List[str] = None
-) -> List[HardwareRequirement]:
-    """Parse the 'requires' field into list of HardwareRequirement.
-
-    Supports:
-    - [all, inactive] -> all hardware + lifecycle nodes must be at that state
-    - [component_name, active] -> specific hardware or lifecycle node must be at that state
-    """
-    if lifecycle_nodes is None:
-        lifecycle_nodes = []
-
-    if not requires:
-        return []
-
-    # we can get either a single [component, state] entry
-    # or a list of [component, state] entries.
-    # here we normalize so we work with a list of [component, state]
-    if len(requires) == 2 and isinstance(requires[0], str):
-        requires_normalized = [requires]
-    else:
-        requires_normalized = requires
-
-    reqs = []
-
-    for req in requires_normalized:
-        if not isinstance(req, list) or len(req) != 2:
-            raise ValueError(f"Invalid requirement format: {req}. Expected [target, state].")
-
-        target = req[0]
-        state = parse_state_string(req[1])
-
-        if target == 'all':
-            reqs.extend([
-                HardwareRequirement(name=name, state=state)
-                for name in hardware + lifecycle_nodes
-            ])
-        else:
-            reqs.append(HardwareRequirement(name=target, state=state))
-
-    return reqs
-
-
 def parse_yaml_file(file_path: Path) -> ParsedScenario:
     """Parse a scenario YAML file into a ParsedScenario object."""
     with open(file_path, 'r') as f:
@@ -98,16 +54,8 @@ def parse_yaml_file(file_path: Path) -> ParsedScenario:
     hardware = data.get('hardware', [])
     lifecycle_nodes = data.get('lifecycle_nodes', [])
 
-    dependency_rules = []
-    controllers = data.get('controllers', {})
-    for ctrl_name, ctrl_config in controllers.items():
-        requires = ctrl_config.get('requires', [])
-        reqs = parse_requires(requires, hardware, lifecycle_nodes)
-
-        dependency_rules.append(ControllerDependencyRule(
-            controller_name=ctrl_name,
-            required_hardware=reqs
-        ))
+    # Dependencies are inferred at runtime from the controller_manager
+    dependency_rules: List[ControllerDependencyRule] = []
 
     goals = {}
     goal_states = data.get('goal_states', {})
@@ -152,8 +100,6 @@ def parse_yaml_file(file_path: Path) -> ParsedScenario:
             metadata[key] = value
 
     tracked_components = set(hardware + lifecycle_nodes)
-    for rule in dependency_rules:
-        tracked_components.add(rule.controller_name)
     for goal in goals.values():
         tracked_components.update(c.name for c in goal.hardware_goals)
         tracked_components.update(c.name for c in goal.controller_goals)

foreman/foreman/planner.py

@@ -3,6 +3,7 @@
 from foreman.types import Component
 from foreman.types import ComponentType
 from foreman.types import ControllerDependencyRule
+from foreman.types import DependencyProvider
 from foreman.types import LifecycleState
 from foreman.types import SystemGoal
 from foreman.types import SystemState
@@ -12,8 +13,21 @@
 class Planner:
     """Plan the next single step towards the lifecycle state goal of the system."""
 
-    def __init__(self, dependency_rules: List[ControllerDependencyRule]):
-        self.rules = {rule.controller_name: rule for rule in dependency_rules}
+    def __init__(self, dependency_rules: List[ControllerDependencyRule] = None,
+                 dependency_provider: Optional[DependencyProvider] = None):
+        self._dependency_provider = dependency_provider
+        self.rules = {rule.controller_name: rule for rule in (dependency_rules or [])}
+
+    def set_dependency_provider(self, dependency_provider: DependencyProvider):
+        """Set a provider for dependency rules, which can be queried at runtime."""
+        self._dependency_provider = dependency_provider
+
+    def get_current_rules(self):
+        """Return the current dependency rules keyed by controller name."""
+        if self._dependency_provider is not None:
+            self.rules = {rule.controller_name: rule
+                          for rule in self._dependency_provider.get_dependency_rules()}
+        return self.rules
 
     def get_next_transition(
         self, current_state: SystemState, goal: SystemGoal
@@ -23,6 +37,9 @@ def get_next_transition(
 
         Priority: C deactivate > HW Down > HW Up > C cleanup > C config > C activate.
         """
+        # Refresh dependency rules from the provider before computing the next transition.
+        self.get_current_rules()
+
         cmds_hw_step_up = []
         cmds_hw_step_down = []
         cmds_ctrl_config = []

foreman/foreman/types.py

@@ -2,7 +2,7 @@
 from dataclasses import field
 from enum import Enum
 from pathlib import Path
-from typing import Dict, List, Optional
+from typing import Dict, List, Optional, Protocol
 
 from lifecycle_msgs.msg import State
 
@@ -136,6 +136,12 @@ class ControllerDependencyRule:
     required_hardware: List[HardwareRequirement]
 
 
+class DependencyProvider(Protocol):
+    """A source of controller dependency rules the planner can pull from."""
+
+    def get_dependency_rules(self) -> List['ControllerDependencyRule']: ...
+
+
 @dataclass
 class SystemGoal:
     """Named system goal state. Populated from YAML."""