fix: improve stuck detection in interactive mode

- Use headless_mode flag to determine stuck detection behavior
- In interactive mode (not headless), only consider history after last user message
- Keep existing behavior in headless mode
- Add comprehensive tests for both modes

Fix: #5480

openhands/controller/agent_controller.py

@@ -902,20 +902,17 @@ def _apply_conversation_window(self, events: list[Event]) -> list[Event]:
 
         return kept_events
 
-    def _is_stuck(self, ui_mode: bool | None = None) -> bool:
+    def _is_stuck(self) -> bool:
         """Checks if the agent or its delegate is stuck in a loop.
 
-        Args:
-            ui_mode: Optional override for UI mode. If not provided, uses not self.headless_mode.
-
         Returns:
             bool: True if the agent is stuck, False otherwise.
         """
         # check if delegate stuck
-        if self.delegate and self.delegate._is_stuck(ui_mode):
+        if self.delegate and self.delegate._is_stuck():
             return True
 
-        return self._stuck_detector.is_stuck(ui_mode if ui_mode is not None else not self.headless_mode)
+        return self._stuck_detector.is_stuck(not self.headless_mode)
 
     def __repr__(self):
         return (

openhands/controller/stuck.py

@@ -24,9 +24,18 @@ class StuckDetector:
     def __init__(self, state: State):
         self.state = state
 
-    def is_stuck(self, ui_mode: bool = False):
-        if ui_mode:
-            # In UI mode, only look at history after the last user message
+    def is_stuck(self, interactive_mode: bool = False):
+        """Checks if the agent is stuck in a loop.
+
+        Args:
+            interactive_mode: If True (not headless), only consider history after last user message.
+                           If False (headless), consider all history.
+
+        Returns:
+            bool: True if the agent is stuck in a loop, False otherwise.
+        """
+        if interactive_mode:
+            # In interactive mode, only look at history after the last user message
             last_user_msg_idx = -1
             for i, event in enumerate(self.state.history):
                 if isinstance(event, MessageAction) and event.source == EventSource.USER:

tests/unit/test_is_stuck.py

@@ -112,10 +112,10 @@ def test_history_too_short(self, stuck_detector: StuckDetector):
         # cmd_observation._cause = cmd_action._id
         state.history.append(cmd_observation)
 
-        assert stuck_detector.is_stuck(ui_mode=False) is False
-        assert stuck_detector.is_stuck(ui_mode=True) is False
+        assert stuck_detector.is_stuck(interactive_mode=False) is False
+        assert stuck_detector.is_stuck(interactive_mode=True) is False
 
-    def test_ui_mode_resets_after_user_message(self, stuck_detector: StuckDetector):
+    def test_interactive_mode_resets_after_user_message(self, stuck_detector: StuckDetector):
         state = stuck_detector.state
 
         # First add some actions that would be stuck in non-UI mode
@@ -128,15 +128,15 @@ def test_ui_mode_resets_after_user_message(self, stuck_detector: StuckDetector):
             state.history.append(cmd_observation)
 
         # In non-UI mode, this should be stuck
-        assert stuck_detector.is_stuck(ui_mode=False) is True
+        assert stuck_detector.is_stuck(interactive_mode=False) is True
 
         # Add a user message
         message_action = MessageAction(content='Hello', wait_for_response=False)
         message_action._source = EventSource.USER
         state.history.append(message_action)
 
-        # In UI mode, this should not be stuck because we ignore history before user message
-        assert stuck_detector.is_stuck(ui_mode=True) is False
+        # In interactive mode, this should not be stuck because we ignore history before user message
+        assert stuck_detector.is_stuck(interactive_mode=True) is False
 
         # Add two more identical actions - still not stuck because we need at least 3
         for i in range(2):
@@ -147,7 +147,7 @@ def test_ui_mode_resets_after_user_message(self, stuck_detector: StuckDetector):
             cmd_observation._cause = cmd_action._id
             state.history.append(cmd_observation)
 
-        assert stuck_detector.is_stuck(ui_mode=True) is False
+        assert stuck_detector.is_stuck(interactive_mode=True) is False
 
         # Add two more identical actions - now it should be stuck
         for i in range(2):
@@ -158,7 +158,7 @@ def test_ui_mode_resets_after_user_message(self, stuck_detector: StuckDetector):
             cmd_observation._cause = cmd_action._id
             state.history.append(cmd_observation)
 
-        assert stuck_detector.is_stuck(ui_mode=True) is True
+        assert stuck_detector.is_stuck(interactive_mode=True) is True
 
     def test_is_stuck_repeating_action_observation(self, stuck_detector: StuckDetector):
         state = stuck_detector.state
@@ -194,7 +194,7 @@ def test_is_stuck_repeating_action_observation(self, stuck_detector: StuckDetect
         state.history.append(message_null_observation)
         # 8 events
 
-        assert stuck_detector.is_stuck(ui_mode=False) is False
+        assert stuck_detector.is_stuck(interactive_mode=False) is False
 
         cmd_action_3 = CmdRunAction(command='ls')
         cmd_action_3._id = 3
@@ -205,7 +205,7 @@ def test_is_stuck_repeating_action_observation(self, stuck_detector: StuckDetect
         # 10 events
 
         assert len(state.history) == 10
-        assert stuck_detector.is_stuck(ui_mode=False) is False
+        assert stuck_detector.is_stuck(interactive_mode=False) is False
 
         cmd_action_4 = CmdRunAction(command='ls')
         cmd_action_4._id = 4
@@ -218,7 +218,7 @@ def test_is_stuck_repeating_action_observation(self, stuck_detector: StuckDetect
         assert len(state.history) == 12
 
         with patch('logging.Logger.warning') as mock_warning:
-            assert stuck_detector.is_stuck(ui_mode=False) is True
+            assert stuck_detector.is_stuck(interactive_mode=False) is True
             mock_warning.assert_called_once_with('Action, Observation loop detected')
 
     def test_is_stuck_repeating_action_error(self, stuck_detector: StuckDetector):
@@ -270,7 +270,7 @@ def test_is_stuck_repeating_action_error(self, stuck_detector: StuckDetector):
         # 12 events
 
         with patch('logging.Logger.warning') as mock_warning:
-            assert stuck_detector.is_stuck(ui_mode=False) is True
+            assert stuck_detector.is_stuck(interactive_mode=False) is True
             mock_warning.assert_called_once_with(
                 'Action, ErrorObservation loop detected'
             )
@@ -284,7 +284,7 @@ def test_is_stuck_invalid_syntax_error(self, stuck_detector: StuckDetector):
         )
 
         with patch('logging.Logger.warning'):
-            assert stuck_detector.is_stuck(ui_mode=False) is True
+            assert stuck_detector.is_stuck(interactive_mode=False) is True
 
     def test_is_not_stuck_invalid_syntax_error_random_lines(
         self, stuck_detector: StuckDetector
@@ -297,7 +297,7 @@ def test_is_not_stuck_invalid_syntax_error_random_lines(
         )
 
         with patch('logging.Logger.warning'):
-            assert stuck_detector.is_stuck(ui_mode=False) is False
+            assert stuck_detector.is_stuck(interactive_mode=False) is False
 
     def test_is_not_stuck_invalid_syntax_error_only_three_incidents(
         self, stuck_detector: StuckDetector
@@ -311,7 +311,7 @@ def test_is_not_stuck_invalid_syntax_error_only_three_incidents(
         )
 
         with patch('logging.Logger.warning'):
-            assert stuck_detector.is_stuck(ui_mode=False) is False
+            assert stuck_detector.is_stuck(interactive_mode=False) is False
 
     def test_is_stuck_incomplete_input_error(self, stuck_detector: StuckDetector):
         state = stuck_detector.state
@@ -322,7 +322,7 @@ def test_is_stuck_incomplete_input_error(self, stuck_detector: StuckDetector):
         )
 
         with patch('logging.Logger.warning'):
-            assert stuck_detector.is_stuck(ui_mode=False) is True
+            assert stuck_detector.is_stuck(interactive_mode=False) is True
 
     def test_is_not_stuck_incomplete_input_error(self, stuck_detector: StuckDetector):
         state = stuck_detector.state
@@ -333,7 +333,7 @@ def test_is_not_stuck_incomplete_input_error(self, stuck_detector: StuckDetector
         )
 
         with patch('logging.Logger.warning'):
-            assert stuck_detector.is_stuck(ui_mode=False) is False
+            assert stuck_detector.is_stuck(interactive_mode=False) is False
 
     def test_is_not_stuck_ipython_unterminated_string_error_random_lines(
         self, stuck_detector: StuckDetector
@@ -342,7 +342,7 @@ def test_is_not_stuck_ipython_unterminated_string_error_random_lines(
         self._impl_unterminated_string_error_events(state, random_line=True)
 
         with patch('logging.Logger.warning'):
-            assert stuck_detector.is_stuck(ui_mode=False) is False
+            assert stuck_detector.is_stuck(interactive_mode=False) is False
 
     def test_is_not_stuck_ipython_unterminated_string_error_only_three_incidents(
         self, stuck_detector: StuckDetector
@@ -353,7 +353,7 @@ def test_is_not_stuck_ipython_unterminated_string_error_only_three_incidents(
         )
 
         with patch('logging.Logger.warning'):
-            assert stuck_detector.is_stuck(ui_mode=False) is False
+            assert stuck_detector.is_stuck(interactive_mode=False) is False
 
     def test_is_stuck_ipython_unterminated_string_error(
         self, stuck_detector: StuckDetector
@@ -362,7 +362,7 @@ def test_is_stuck_ipython_unterminated_string_error(
         self._impl_unterminated_string_error_events(state, random_line=False)
 
         with patch('logging.Logger.warning'):
-            assert stuck_detector.is_stuck(ui_mode=False) is True
+            assert stuck_detector.is_stuck(interactive_mode=False) is True
 
     def test_is_not_stuck_ipython_syntax_error_not_at_end(
         self, stuck_detector: StuckDetector
@@ -407,7 +407,7 @@ def test_is_not_stuck_ipython_syntax_error_not_at_end(
         state.history.append(ipython_observation_4)
 
         with patch('logging.Logger.warning') as mock_warning:
-            assert stuck_detector.is_stuck(ui_mode=False) is False
+            assert stuck_detector.is_stuck(interactive_mode=False) is False
             mock_warning.assert_not_called()
 
     def test_is_stuck_repeating_action_observation_pattern(
@@ -476,7 +476,7 @@ def test_is_stuck_repeating_action_observation_pattern(
         state.history.append(read_observation_3)
 
         with patch('logging.Logger.warning') as mock_warning:
-            assert stuck_detector.is_stuck(ui_mode=False) is True
+            assert stuck_detector.is_stuck(interactive_mode=False) is True
             mock_warning.assert_called_once_with('Action, Observation pattern detected')
 
     def test_is_stuck_not_stuck(self, stuck_detector: StuckDetector):
@@ -542,7 +542,7 @@ def test_is_stuck_not_stuck(self, stuck_detector: StuckDetector):
         # read_observation_3._cause = read_action_3._id
         state.history.append(read_observation_3)
 
-        assert stuck_detector.is_stuck(ui_mode=False) is False
+        assert stuck_detector.is_stuck(interactive_mode=False) is False
 
     def test_is_stuck_monologue(self, stuck_detector):
         state = stuck_detector.state
@@ -572,7 +572,7 @@ def test_is_stuck_monologue(self, stuck_detector):
         message_action_6._source = EventSource.AGENT
         state.history.append(message_action_6)
 
-        assert stuck_detector.is_stuck(ui_mode=False)
+        assert stuck_detector.is_stuck(interactive_mode=False)
 
         # Add an observation event between the repeated message actions
         cmd_output_observation = CmdOutputObservation(
@@ -592,7 +592,7 @@ def test_is_stuck_monologue(self, stuck_detector):
         state.history.append(message_action_8)
 
         with patch('logging.Logger.warning'):
-            assert not stuck_detector.is_stuck(ui_mode=False)
+            assert not stuck_detector.is_stuck(interactive_mode=False)
 
 
 class TestAgentController:
@@ -609,4 +609,4 @@ def controller(self):
     def test_is_stuck_delegate_stuck(self, controller: AgentController):
         controller.delegate = Mock()
         controller.delegate._is_stuck.return_value = True
-        assert controller._is_stuck(ui_mode=False) is True
+        assert controller._is_stuck() is True