diff --git a/examples/20_stuck_detector.py b/examples/20_stuck_detector.py
new file mode 100644
index 0000000000..3c6e51be42
--- /dev/null
+++ b/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]}")
diff --git a/openhands/sdk/conversation/__init__.py b/openhands/sdk/conversation/__init__.py
index 6dafefaa57..0dbc29b008 100644
--- a/openhands/sdk/conversation/__init__.py
+++ b/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",
 ]
diff --git a/openhands/sdk/conversation/conversation.py b/openhands/sdk/conversation/conversation.py
index d86e75df63..807b233673 100644
--- a/openhands/sdk/conversation/conversation.py
+++ b/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
diff --git a/openhands/sdk/conversation/state.py b/openhands/sdk/conversation/state.py
index c30cb89181..a5b292682f 100644
--- a/openhands/sdk/conversation/state.py
+++ b/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:
diff --git a/openhands/sdk/conversation/stuck_detector.py b/openhands/sdk/conversation/stuck_detector.py
new file mode 100644
index 0000000000..baf5d88416
--- /dev/null
+++ b/openhands/sdk/conversation/stuck_detector.py
@@ -0,0 +1,275 @@
+from openhands.sdk.conversation.state import ConversationState
+from openhands.sdk.event import (
+    ActionEvent,
+    AgentErrorEvent,
+    CondensationSummaryEvent,
+    EventBase,
+    MessageEvent,
+    ObservationEvent,
+)
+from openhands.sdk.logger import get_logger
+
+
+logger = get_logger(__name__)
+
+
+class StuckDetector:
+    """Detects when an agent is stuck in repetitive or unproductive patterns.
+
+    This detector analyzes the conversation history to identify various stuck patterns:
+    1. Repeating action-observation cycles
+    2. Repeating action-error cycles
+    3. Agent monologue (repeated messages without user input)
+    4. Repeating alternating action-observation patterns
+    5. Context window errors indicating memory issues
+    """
+
+    def __init__(self, state: ConversationState):
+        self.state = state
+
+    def is_stuck(self) -> bool:
+        """Check if the agent is currently stuck."""
+        events = list(self.state.events)
+
+        # Only look at history after the last user message
+        last_user_msg_index = next(
+            (
+                i
+                for i in reversed(range(len(events)))
+                if isinstance(events[i], MessageEvent) and events[i].source == "user"
+            ),
+            -1,  # Default to -1 if no user message found
+        )
+        if last_user_msg_index == -1:
+            logger.warning("No user message found in history, skipping stuck detection")
+            return False
+
+        events = events[last_user_msg_index + 1 :]
+
+        # it takes 3 actions minimum to detect a loop, otherwise nothing to do here
+        if len(events) < 3:
+            return False
+
+        logger.debug(f"Checking for stuck patterns in {len(events)} events")
+        logger.debug(
+            f"Events after last user message: {[type(e).__name__ for e in events]}"
+        )
+
+        # the first few scenarios detect 3 or 4 repeated steps
+        # prepare the last 4 actions and observations, to check them out
+        last_actions: list[EventBase] = []
+        last_observations: list[EventBase] = []
+
+        # retrieve the last four actions and observations starting from
+        # the end of history, wherever they are
+        for event in reversed(events):
+            if isinstance(event, ActionEvent) and len(last_actions) < 4:
+                last_actions.append(event)
+            elif (
+                isinstance(event, (ObservationEvent, AgentErrorEvent))
+                and len(last_observations) < 4
+            ):
+                last_observations.append(event)
+            if len(last_actions) >= 4 and len(last_observations) >= 4:
+                break
+
+        # Check all stuck patterns
+        # scenario 1: same action, same observation
+        if self._is_stuck_repeating_action_observation(last_actions, last_observations):
+            return True
+
+        # scenario 2: same action, errors
+        if self._is_stuck_repeating_action_error(last_actions, last_observations):
+            return True
+
+        # scenario 3: monologue
+        if self._is_stuck_monologue(events):
+            return True
+
+        # scenario 4: action, observation alternating pattern on the last six steps
+        if len(events) >= 6:
+            if self._is_stuck_alternating_action_observation(events):
+                return True
+
+        # scenario 5: context window error loop
+        if len(events) >= 10:
+            if self._is_stuck_context_window_error(events):
+                return True
+
+        return False
+
+    def _is_stuck_repeating_action_observation(
+        self, last_actions: list[EventBase], last_observations: list[EventBase]
+    ) -> bool:
+        # scenario 1: same action, same observation
+        # it takes 4 actions and 4 observations to detect a loop
+        # assert len(last_actions) == 4 and len(last_observations) == 4
+
+        # Check for a loop of 4 identical action-observation pairs
+        if len(last_actions) == 4 and len(last_observations) == 4:
+            logger.debug("Found 4 actions and 4 observations, checking for equality")
+            actions_equal = all(
+                self._event_eq(last_actions[0], action) for action in last_actions
+            )
+            observations_equal = all(
+                self._event_eq(last_observations[0], observation)
+                for observation in last_observations
+            )
+            logger.debug(
+                f"Actions equal: {actions_equal}, "
+                f"Observations equal: {observations_equal}"
+            )
+
+            if actions_equal and observations_equal:
+                logger.warning("Action, Observation loop detected")
+                return True
+        else:
+            logger.debug(
+                f"Not enough actions/observations: {len(last_actions)} actions,"
+                f" {len(last_observations)} observations"
+            )
+
+        return False
+
+    def _is_stuck_repeating_action_error(
+        self, last_actions: list[EventBase], last_observations: list[EventBase]
+    ) -> bool:
+        # scenario 2: same action, errors
+        # it takes 3 actions and 3 observations to detect a loop
+        # check if the last three actions are the same and result in errors
+        if len(last_actions) < 3 or len(last_observations) < 3:
+            return False
+
+        # are the last three actions the "same"?
+        if all(self._event_eq(last_actions[0], action) for action in last_actions[:3]):
+            # and the last three observations are all errors?
+            if all(isinstance(obs, AgentErrorEvent) for obs in last_observations[:3]):
+                logger.warning("Action, Error loop detected")
+                return True
+
+        # Check if observations are errors
+        return False
+
+    def _is_stuck_monologue(self, events: list[EventBase]) -> bool:
+        # scenario 3: monologue
+        # check for repeated MessageActions with source=AGENT
+        # see if the agent is engaged in a good old monologue, telling
+        # itself the same thing over and over
+        if len(events) < 3:
+            return False
+
+        # Look for 3 consecutive agent messages without user interruption
+        agent_message_count = 0
+
+        for event in reversed(events):
+            if isinstance(event, MessageEvent):
+                if event.source == "agent":
+                    agent_message_count += 1
+                elif event.source == "user":
+                    break  # User interrupted, not a monologue
+            elif isinstance(event, CondensationSummaryEvent):
+                # Condensation events don't break the monologue pattern
+                continue
+            else:
+                # Other events (actions/observations) don't count as monologue
+                break
+
+        return agent_message_count >= 3
+
+    def _is_stuck_alternating_action_observation(self, events: list[EventBase]) -> bool:
+        # scenario 4: alternating action-observation loop
+        # needs 6 actions and 6 observations to detect the ping-pong pattern
+
+        last_actions: list[EventBase] = []
+        last_observations: list[EventBase] = []
+
+        # collect most recent 6 actions and 6 observations
+        for event in reversed(events):
+            if isinstance(event, ActionEvent) and len(last_actions) < 6:
+                last_actions.append(event)
+            elif (
+                isinstance(event, (ObservationEvent, AgentErrorEvent))
+                and len(last_observations) < 6
+            ):
+                last_observations.append(event)
+
+            if len(last_actions) == 6 and len(last_observations) == 6:
+                break
+
+        if len(last_actions) == 6 and len(last_observations) == 6:
+            actions_equal = (
+                self._event_eq(last_actions[0], last_actions[2])
+                and self._event_eq(last_actions[0], last_actions[4])
+                and self._event_eq(last_actions[1], last_actions[3])
+                and self._event_eq(last_actions[1], last_actions[5])
+            )
+            observations_equal = (
+                self._event_eq(last_observations[0], last_observations[2])
+                and self._event_eq(last_observations[0], last_observations[4])
+                and self._event_eq(last_observations[1], last_observations[3])
+                and self._event_eq(last_observations[1], last_observations[5])
+            )
+
+            if actions_equal and observations_equal:
+                logger.warning("Alternating Action, Observation loop detected")
+                return True
+
+        return False
+
+    def _is_stuck_context_window_error(self, events: list[EventBase]) -> bool:
+        """Detects if we're stuck in a loop of context window errors.
+
+        This happens when we repeatedly get context window errors and try to trim,
+        but the trimming doesn't work, causing us to get more context window errors.
+        The pattern is repeated AgentCondensationObservation events without any other
+        events between them.
+        """
+        # TODO: blocked by https://github.com/All-Hands-AI/agent-sdk/issues/282
+        return False
+
+    def _event_eq(self, event1: EventBase, event2: EventBase) -> bool:
+        """
+        Compare two events for equality, ignoring irrelevant
+        details like ids, metrics.
+        """
+        # Must be same type
+        if type(event1) is not type(event2):
+            return False
+
+        # For ActionEvents, compare the action content, ignoring IDs
+        if isinstance(event1, ActionEvent) and isinstance(event2, ActionEvent):
+            return (
+                event1.source == event2.source
+                and event1.thought == event2.thought
+                and event1.action == event2.action
+                and event1.tool_name == event2.tool_name
+                # Ignore tool_call_id, llm_response_id, action_id as they vary
+            )
+
+        # For ObservationEvents, compare the observation content, ignoring IDs
+        if isinstance(event1, ObservationEvent) and isinstance(
+            event2, ObservationEvent
+        ):
+            return (
+                event1.source == event2.source
+                and event1.observation == event2.observation
+                and event1.tool_name == event2.tool_name
+                # Ignore action_id, tool_call_id as they vary
+            )
+
+        # For AgentErrorEvents, compare the error content
+        if isinstance(event1, AgentErrorEvent) and isinstance(event2, AgentErrorEvent):
+            return (
+                event1.source == event2.source and event1.error == event2.error
+                # Ignore action_id as it varies
+            )
+
+        # For MessageEvents, compare the message content
+        if isinstance(event1, MessageEvent) and isinstance(event2, MessageEvent):
+            return (
+                event1.source == event2.source
+                and event1.llm_message == event2.llm_message
+            )
+
+        # Default fallback
+        return event1 == event2
diff --git a/tests/cross/test_stuck_detector.py b/tests/cross/test_stuck_detector.py
new file mode 100644
index 0000000000..f67150d37d
--- /dev/null
+++ b/tests/cross/test_stuck_detector.py
@@ -0,0 +1,380 @@
+import uuid
+
+from litellm import ChatCompletionMessageToolCall
+
+from openhands.sdk.agent import Agent
+from openhands.sdk.conversation.state import ConversationState
+from openhands.sdk.conversation.stuck_detector import StuckDetector
+from openhands.sdk.event import (
+    ActionEvent,
+    AgentErrorEvent,
+    MessageEvent,
+    ObservationEvent,
+)
+from openhands.sdk.llm import LLM, Message, TextContent
+from openhands.tools.execute_bash.definition import (
+    ExecuteBashAction,
+    ExecuteBashObservation,
+)
+
+
+def test_history_too_short():
+    """Test that stuck detector returns False when there are too few events."""
+    # Create a minimal agent for testing
+    llm = LLM(model="gpt-4o-mini")
+    agent = Agent(llm=llm)
+    state = ConversationState.create(id=uuid.uuid4(), agent=agent)
+    stuck_detector = StuckDetector(state)
+
+    # Add a user message
+    user_message = MessageEvent(
+        source="user",
+        llm_message=Message(role="user", content=[TextContent(text="Hello")]),
+    )
+    state.events.append(user_message)
+
+    # Add a single action-observation pair
+    action = ActionEvent(
+        source="agent",
+        thought=[TextContent(text="I need to run ls command")],
+        action=ExecuteBashAction(command="ls"),
+        tool_name="execute_bash",
+        tool_call_id="call_1",
+        tool_call=ChatCompletionMessageToolCall(
+            id="call_1",
+            function={"name": "execute_bash", "arguments": '{"command": "ls"}'},
+            type="function",
+        ),
+        llm_response_id="response_1",
+    )
+    state.events.append(action)
+
+    observation = ObservationEvent(
+        source="environment",
+        observation=ExecuteBashObservation(
+            output="file1.txt\nfile2.txt", command="ls", exit_code=0
+        ),
+        action_id=action.id,
+        tool_name="execute_bash",
+        tool_call_id="call_1",
+    )
+    state.events.append(observation)
+
+    # Should not be stuck with only one action-observation pair after user message
+    assert stuck_detector.is_stuck() is False
+
+
+def test_repeating_action_observation_not_stuck_less_than_4_repeats():
+    """Test detection of repeating action-observation cycles."""
+    llm = LLM(model="gpt-4o-mini")
+    agent = Agent(llm=llm)
+    state = ConversationState.create(id=uuid.uuid4(), agent=agent)
+    stuck_detector = StuckDetector(state)
+
+    # Add a user message first
+    user_message = MessageEvent(
+        source="user",
+        llm_message=Message(role="user", content=[TextContent(text="Please run ls")]),
+    )
+    state.events.append(user_message)
+
+    # Add 3 identical action-observation pairs to trigger stuck detection
+    for i in range(3):
+        action = ActionEvent(
+            source="agent",
+            thought=[TextContent(text="I need to run ls command")],
+            action=ExecuteBashAction(command="ls"),
+            tool_name="execute_bash",
+            tool_call_id=f"call_{i}",
+            tool_call=ChatCompletionMessageToolCall(
+                id=f"call_{i}",
+                function={"name": "execute_bash", "arguments": '{"command": "ls"}'},
+                type="function",
+            ),
+            llm_response_id=f"response_{i}",
+        )
+        state.events.append(action)
+
+        observation = ObservationEvent(
+            source="environment",
+            observation=ExecuteBashObservation(
+                output="file1.txt\nfile2.txt", command="ls", exit_code=0
+            ),
+            action_id=action.id,
+            tool_name="execute_bash",
+            tool_call_id=f"call_{i}",
+        )
+        state.events.append(observation)
+
+    # Should be stuck with 4 identical action-observation pairs
+    assert stuck_detector.is_stuck() is False
+
+
+def test_repeating_action_observation_stuck():
+    """Test detection of repeating action-observation cycles."""
+    llm = LLM(model="gpt-4o-mini")
+    agent = Agent(llm=llm)
+    state = ConversationState.create(id=uuid.uuid4(), agent=agent)
+    stuck_detector = StuckDetector(state)
+
+    # Add a user message first
+    user_message = MessageEvent(
+        source="user",
+        llm_message=Message(role="user", content=[TextContent(text="Please run ls")]),
+    )
+    state.events.append(user_message)
+
+    # Add 4 identical action-observation pairs to trigger stuck detection
+    for i in range(4):
+        action = ActionEvent(
+            source="agent",
+            thought=[TextContent(text="I need to run ls command")],
+            action=ExecuteBashAction(command="ls"),
+            tool_name="execute_bash",
+            tool_call_id=f"call_{i}",
+            tool_call=ChatCompletionMessageToolCall(
+                id=f"call_{i}",
+                function={"name": "execute_bash", "arguments": '{"command": "ls"}'},
+                type="function",
+            ),
+            llm_response_id=f"response_{i}",
+        )
+        state.events.append(action)
+
+        observation = ObservationEvent(
+            source="environment",
+            observation=ExecuteBashObservation(
+                output="file1.txt\nfile2.txt", command="ls", exit_code=0
+            ),
+            action_id=action.id,
+            tool_name="execute_bash",
+            tool_call_id=f"call_{i}",
+        )
+        state.events.append(observation)
+
+    # Should be stuck with 4 identical action-observation pairs
+    assert stuck_detector.is_stuck() is True
+
+
+def test_repeating_action_error_stuck():
+    """Test detection of repeating action-error cycles."""
+    llm = LLM(model="gpt-4o-mini")
+    agent = Agent(llm=llm)
+    state = ConversationState.create(id=uuid.uuid4(), agent=agent)
+    stuck_detector = StuckDetector(state)
+
+    # Add a user message first
+    user_message = MessageEvent(
+        source="user",
+        llm_message=Message(
+            role="user", content=[TextContent(text="Please run the invalid command")]
+        ),
+    )
+    state.events.append(user_message)
+
+    def create_action_and_error(i):
+        action = ActionEvent(
+            source="agent",
+            thought=[TextContent(text="I need to run invalid_command")],
+            action=ExecuteBashAction(command="invalid_command"),
+            tool_name="execute_bash",
+            tool_call_id=f"call_{i}",
+            tool_call=ChatCompletionMessageToolCall(
+                id=f"call_{i}",
+                function={
+                    "name": "execute_bash",
+                    "arguments": '{"command": "invalid_command"}',
+                },
+                type="function",
+            ),
+            llm_response_id=f"response_{i}",
+        )
+        error = AgentErrorEvent(
+            source="agent", error="Command 'invalid_command' not found"
+        )
+        return action, error
+
+    # Add 2 identical actions that result in errors
+    for i in range(2):
+        action, error = create_action_and_error(i)
+        state.events.append(action)
+        state.events.append(error)
+
+    # Should not stuck with 2 identical action-error pairs
+    assert stuck_detector.is_stuck() is False
+
+    # Add 1 more identical action-error pair to trigger stuck detection
+    action, error = create_action_and_error(2)
+    state.events.append(action)
+    state.events.append(error)
+
+    # Should be stuck with 3 identical action-error pairs
+    assert stuck_detector.is_stuck() is True
+
+
+def test_agent_monologue_stuck():
+    """Test detection of agent monologue (repeated messages without user input)."""
+    llm = LLM(model="gpt-4o-mini")
+    agent = Agent(llm=llm)
+    state = ConversationState.create(id=uuid.uuid4(), agent=agent)
+    stuck_detector = StuckDetector(state)
+
+    # Add a user message first
+    user_message = MessageEvent(
+        source="user",
+        llm_message=Message(role="user", content=[TextContent(text="Hello")]),
+    )
+    state.events.append(user_message)
+
+    # Add 3 consecutive agent messages (monologue)
+    for i in range(3):
+        agent_message = MessageEvent(
+            source="agent",
+            llm_message=Message(
+                role="assistant", content=[TextContent(text=f"I'm thinking... {i}")]
+            ),
+        )
+        state.events.append(agent_message)
+
+    # Should be stuck due to agent monologue
+    assert stuck_detector.is_stuck() is True
+
+
+def test_not_stuck_with_different_actions():
+    """Test that different actions don't trigger stuck detection."""
+    llm = LLM(model="gpt-4o-mini")
+    agent = Agent(llm=llm)
+    state = ConversationState.create(id=uuid.uuid4(), agent=agent)
+    stuck_detector = StuckDetector(state)
+
+    # Add a user message first
+    user_message = MessageEvent(
+        source="user",
+        llm_message=Message(
+            role="user", content=[TextContent(text="Please run different commands")]
+        ),
+    )
+    state.events.append(user_message)
+
+    # Add different actions
+    commands = ["ls", "pwd", "whoami", "date"]
+    for i, cmd in enumerate(commands):
+        action = ActionEvent(
+            source="agent",
+            thought=[TextContent(text=f"I need to run {cmd} command")],
+            action=ExecuteBashAction(command=cmd),
+            tool_name="execute_bash",
+            tool_call_id=f"call_{i}",
+            tool_call=ChatCompletionMessageToolCall(
+                id=f"call_{i}",
+                function={
+                    "name": "execute_bash",
+                    "arguments": f'{{"command": "{cmd}"}}',
+                },
+                type="function",
+            ),
+            llm_response_id=f"response_{i}",
+        )
+        state.events.append(action)
+
+        observation = ObservationEvent(
+            source="environment",
+            observation=ExecuteBashObservation(
+                output=f"output from {cmd}", command=cmd, exit_code=0
+            ),
+            action_id=action.id,
+            tool_name="execute_bash",
+            tool_call_id=f"call_{i}",
+        )
+        state.events.append(observation)
+
+    # Should not be stuck with different actions
+    assert stuck_detector.is_stuck() is False
+
+
+def test_reset_after_user_message():
+    """Test that stuck detection resets after a new user message."""
+    llm = LLM(model="gpt-4o-mini")
+    agent = Agent(llm=llm)
+    state = ConversationState.create(id=uuid.uuid4(), agent=agent)
+    stuck_detector = StuckDetector(state)
+
+    # Add initial user message
+    user_message = MessageEvent(
+        source="user",
+        llm_message=Message(role="user", content=[TextContent(text="Please run ls")]),
+    )
+    state.events.append(user_message)
+
+    # Add 4 identical action-observation pairs to trigger stuck detection
+    for i in range(4):
+        action = ActionEvent(
+            source="agent",
+            thought=[TextContent(text="I need to run ls command")],
+            action=ExecuteBashAction(command="ls"),
+            tool_name="execute_bash",
+            tool_call_id=f"call_{i}",
+            tool_call=ChatCompletionMessageToolCall(
+                id=f"call_{i}",
+                function={"name": "execute_bash", "arguments": '{"command": "ls"}'},
+                type="function",
+            ),
+            llm_response_id=f"response_{i}",
+        )
+        state.events.append(action)
+
+        observation = ObservationEvent(
+            source="environment",
+            observation=ExecuteBashObservation(
+                output="file1.txt\nfile2.txt", command="ls", exit_code=0
+            ),
+            action_id=action.id,
+            tool_name="execute_bash",
+            tool_call_id=f"call_{i}",
+        )
+        state.events.append(observation)
+
+    # Should be stuck
+    assert stuck_detector.is_stuck() is True
+
+    # Add a new user message
+    new_user_message = MessageEvent(
+        source="user",
+        llm_message=Message(
+            role="user", content=[TextContent(text="Try something else")]
+        ),
+    )
+    state.events.append(new_user_message)
+
+    # Should not be stuck after new user message (history is reset)
+    assert stuck_detector.is_stuck() is False
+
+    # Add one more action after user message - still not stuck
+    action = ActionEvent(
+        source="agent",
+        thought=[TextContent(text="I'll try pwd command")],
+        action=ExecuteBashAction(command="pwd"),
+        tool_name="execute_bash",
+        tool_call_id="call_new",
+        tool_call=ChatCompletionMessageToolCall(
+            id="call_new",
+            function={"name": "execute_bash", "arguments": '{"command": "pwd"}'},
+            type="function",
+        ),
+        llm_response_id="response_new",
+    )
+    state.events.append(action)
+
+    observation = ObservationEvent(
+        source="environment",
+        observation=ExecuteBashObservation(
+            output="/home/user", command="pwd", exit_code=0
+        ),
+        action_id=action.id,
+        tool_name="execute_bash",
+        tool_call_id="call_new",
+    )
+    state.events.append(observation)
+
+    # Still not stuck with just one action after user message
+    assert stuck_detector.is_stuck() is False
diff --git a/tests/sdk/conversation/test_pause_functionality.py b/tests/sdk/conversation/test_pause_functionality.py
index 3d3f60bf52..8cdf05411d 100644
--- a/tests/sdk/conversation/test_pause_functionality.py
+++ b/tests/sdk/conversation/test_pause_functionality.py
@@ -300,7 +300,7 @@ def _make_blocking_tool() -> Tool:
             llm=self.llm,
             tools=[ToolSpec(name="test_tool")],
         )
-        conversation = Conversation(agent=agent)
+        conversation = Conversation(agent=agent, stuck_detection=False)
 
         # Swap them in for this test only
         self.agent = agent