Port over stuck detector

examples/20_stuck_detector.py

@@ -0,0 +1,60 @@
+import os
+
+from pydantic import SecretStr
+
+from openhands.sdk import (
+    LLM,
+    Conversation,
+    EventBase,
+    LLMConvertibleEvent,
+    get_logger,
+)
+from openhands.sdk.preset.default import get_default_agent
+
+
+logger = get_logger(__name__)
+
+# Configure LLM
+api_key = os.getenv("LITELLM_API_KEY")
+assert api_key is not None, "LITELLM_API_KEY environment variable is not set."
+llm = LLM(
+    model="litellm_proxy/anthropic/claude-sonnet-4-20250514",
+    base_url="https://llm-proxy.eval.all-hands.dev",
+    api_key=SecretStr(api_key),
+)
+
+agent = get_default_agent(llm=llm, working_dir=os.getcwd())
+
+llm_messages = []
+
+
+def conversation_callback(event: EventBase):
+    if isinstance(event, LLMConvertibleEvent):
+        llm_messages.append(event.to_llm_message())
+
+
+# Create conversation with built-in stuck detection
+conversation = Conversation(
+    agent=agent,
+    callbacks=[conversation_callback],
+    # This is by default True, shown here for clarity of the example
+    stuck_detection=True,
+)
+
+# Send a task that will be caught by stuck detection
+conversation.send_message(
+    "Please execute 'ls' command 5 times, each in its own "
+    "action without any thought and then exit at the 6th step."
+)
+
+# Run the conversation - stuck detection happens automatically
+conversation.run()
+
+assert conversation.stuck_detector is not None
+final_stuck_check = conversation.stuck_detector.is_stuck()
+print(f"Final stuck status: {final_stuck_check}")
+
+print("=" * 100)
+print("Conversation finished. Got the following LLM messages:")
+for i, message in enumerate(llm_messages):
+    print(f"Message {i}: {str(message)[:200]}")

openhands/sdk/conversation/__init__.py

@@ -2,6 +2,7 @@
 from openhands.sdk.conversation.event_store import EventLog, ListLike
 from openhands.sdk.conversation.secrets_manager import SecretsManager
 from openhands.sdk.conversation.state import ConversationState
+from openhands.sdk.conversation.stuck_detector import StuckDetector
 from openhands.sdk.conversation.types import ConversationCallbackType
 from openhands.sdk.conversation.visualizer import ConversationVisualizer
 
@@ -12,6 +13,7 @@
     "ConversationCallbackType",
     "ConversationVisualizer",
     "SecretsManager",
+    "StuckDetector",
     "EventLog",
     "ListLike",
 ]

openhands/sdk/conversation/conversation.py

@@ -4,6 +4,7 @@
 from openhands.sdk.agent.base import AgentBase
 from openhands.sdk.conversation.secrets_manager import SecretValue
 from openhands.sdk.conversation.state import AgentExecutionStatus, ConversationState
+from openhands.sdk.conversation.stuck_detector import StuckDetector
 from openhands.sdk.conversation.types import ConversationCallbackType, ConversationID
 from openhands.sdk.conversation.visualizer import (
     create_default_visualizer,
@@ -43,6 +44,7 @@ def __init__(
         callbacks: list[ConversationCallbackType] | None = None,
         max_iteration_per_run: int = 500,
         visualize: bool = True,
+        stuck_detection: bool = True,
     ):
         """Initialize the conversation.
 
@@ -57,6 +59,7 @@ def __init__(
             visualize: Whether to enable default visualization. If True, adds
                       a default visualizer callback. If False, relies on
                       application to provide visualization through callbacks.
+            stuck_detection: Whether to enable stuck detection
         """
         self.agent = agent
         self._persist_filestore = persist_filestore
@@ -85,6 +88,9 @@ def _default_callback(e):
         self._on_event = compose_callbacks(composed_list)
         self.max_iteration_per_run = max_iteration_per_run
 
+        # Initialize stuck detector
+        self._stuck_detector = StuckDetector(self.state) if stuck_detection else None
+
         with self.state:
             self.agent.init_state(self.state, on_event=self._on_event)
 
@@ -93,6 +99,11 @@ def id(self) -> ConversationID:
         """Get the unique ID of the conversation."""
         return self.state.id
 
+    @property
+    def stuck_detector(self) -> StuckDetector | None:
+        """Get the stuck detector instance if enabled."""
+        return self._stuck_detector
+
     def send_message(self, message: str | Message) -> None:
         """Send a message to the agent.
 
@@ -173,9 +184,19 @@ def run(self) -> None:
                 if self.state.agent_status in [
                     AgentExecutionStatus.FINISHED,
                     AgentExecutionStatus.PAUSED,
+                    AgentExecutionStatus.STUCK,
                 ]:
                     break
 
+                # Check for stuck patterns if enabled
+                if self._stuck_detector:
+                    is_stuck = self._stuck_detector.is_stuck()
+
+                    if is_stuck:
+                        logger.warning("Stuck pattern detected.")
+                        self.state.agent_status = AgentExecutionStatus.STUCK
+                        continue
+
                 # clear the flag before calling agent.step() (user approved)
                 if (
                     self.state.agent_status

openhands/sdk/conversation/state.py

@@ -36,6 +36,7 @@ class AgentExecutionStatus(str, Enum):
     )
     FINISHED = "finished"  # Agent has completed the current task
     ERROR = "error"  # Agent encountered an error (optional for future use)
+    STUCK = "stuck"  # Agent is stuck in a loop or unable to proceed
 
 
 if TYPE_CHECKING: