From ae36b6e196dfae89bf012ee1097d65f46d0cb403 Mon Sep 17 00:00:00 2001 From: Marc Mintel Date: Mon, 27 Apr 2026 15:24:10 +0200 Subject: [PATCH] fix(goap): resolve infinite routing loop by feeding masked actions to HD Map pathfinder --- GramAddict/core/goap.py | 2 +- GramAddict/core/navigation/planner.py | 16 ++++++++++++---- GramAddict/core/screen_topology.py | 7 ++++++- 3 files changed, 19 insertions(+), 6 deletions(-) diff --git a/GramAddict/core/goap.py b/GramAddict/core/goap.py index c10addf..de4abed 100644 --- a/GramAddict/core/goap.py +++ b/GramAddict/core/goap.py @@ -173,7 +173,7 @@ class GoalExecutor: continue # PLAN - action = self.planner.plan_next_step(goal, screen, explored_nav_actions=explored_nav_actions) + action = self.planner.plan_next_step(goal, screen, explored_nav_actions=explored_nav_actions, action_failures=self.action_failures) if action is None: # Goal achieved! diff --git a/GramAddict/core/navigation/planner.py b/GramAddict/core/navigation/planner.py index dad1687..cc8bdd5 100644 --- a/GramAddict/core/navigation/planner.py +++ b/GramAddict/core/navigation/planner.py @@ -17,7 +17,7 @@ class GoalPlanner: def __init__(self, username: str): self.knowledge = NavigationKnowledge(username) - def plan_next_step(self, goal: str, screen: Dict[str, Any], explored_nav_actions: set = None) -> Optional[str]: + def plan_next_step(self, goal: str, screen: Dict[str, Any], explored_nav_actions: set = None, action_failures: dict = None) -> Optional[str]: """Plans the NEXT single action to take toward the goal.""" screen_type = screen["screen_type"] available = screen.get("available_actions", []) @@ -34,7 +34,7 @@ class GoalPlanner: # ── 3. Am I on the right screen? If not, navigate there ── selected_tab = screen.get("selected_tab") - nav_action = self._plan_navigation(goal_lower, screen_type, available, selected_tab, explored_nav_actions) + nav_action = self._plan_navigation(goal_lower, screen_type, available, selected_tab, explored_nav_actions, action_failures) if nav_action: return nav_action @@ -70,6 +70,7 @@ class GoalPlanner: available: List[str], selected_tab: Optional[str] = None, explored_nav_actions: set = None, + action_failures: dict = None, ) -> Optional[str]: """If we're on the wrong screen, figure out how to navigate. @@ -88,11 +89,18 @@ class GoalPlanner: else: logger.debug(f"🛡️ [Aversive Filter] Masking trapped action: '{action}'") available = safe_available + + # Build avoid_actions for HD Map route planning + avoid_actions = (explored_nav_actions or set()).copy() + if action_failures: + for act, count in action_failures.items(): + if count >= 2: # MAX_RETRIES is 2 in goap + avoid_actions.add(act) # ── 1. HD Map Routing (Primary Strategy) ── target_screen = ScreenTopology.goal_to_target_screen(goal) if target_screen and target_screen != screen_type: - route = ScreenTopology.find_route(screen_type, target_screen) + route = ScreenTopology.find_route(screen_type, target_screen, avoid_actions=avoid_actions) if route: next_action, next_screen = route[0] # Verify action isn't explored/trapped @@ -131,7 +139,7 @@ class GoalPlanner: # 5. Find the action we need to take (from learned knowledge or HD map) for target_screen in required_screens: # Try HD Map first! - route = ScreenTopology.find_route(screen_type, target_screen) + route = ScreenTopology.find_route(screen_type, target_screen, avoid_actions=avoid_actions) if route: next_action, next_screen = route[0] if next_action not in (explored_nav_actions or set()): diff --git a/GramAddict/core/screen_topology.py b/GramAddict/core/screen_topology.py index cf2d6d1..032d185 100644 --- a/GramAddict/core/screen_topology.py +++ b/GramAddict/core/screen_topology.py @@ -88,7 +88,7 @@ class ScreenTopology: } @classmethod - def find_route(cls, from_screen: ScreenType, to_screen: ScreenType) -> Optional[List[Tuple[str, ScreenType]]]: + def find_route(cls, from_screen: ScreenType, to_screen: ScreenType, avoid_actions: set = None) -> Optional[List[Tuple[str, ScreenType]]]: """ BFS shortest path from from_screen to to_screen. @@ -99,6 +99,8 @@ class ScreenTopology: """ if from_screen == to_screen: return [] + + avoid_actions = avoid_actions or set() queue: deque = deque() queue.append((from_screen, [])) @@ -109,6 +111,9 @@ class ScreenTopology: transitions = cls.TRANSITIONS.get(current, {}) for action, next_screen in transitions.items(): + if action in avoid_actions or action.replace(" ", "_") in avoid_actions: + continue + if next_screen == to_screen: return path + [(action, next_screen)]