Refactor of error handling

evaluation/EDA/run_infer.py

@@ -158,7 +158,7 @@ def process_instance(
         metadata=metadata,
         history=histories,
         metrics=metrics,
-        error=state.last_error if state and state.last_error else None,
+        error=state.get_last_error() if state and state.get_last_error() else None,
         test_result={
             'success': test_result,
             'final_message': final_message,

evaluation/agent_bench/run_infer.py

@@ -283,7 +283,7 @@ def process_instance(
         metadata=metadata,
         history=histories,
         metrics=metrics,
-        error=state.last_error if state and state.last_error else None,
+        error=state.get_last_error() if state and state.get_last_error() else None,
         test_result={
             'agent_answer': agent_answer,
             'final_answer': final_ans,

evaluation/aider_bench/run_infer.py

@@ -261,7 +261,7 @@ def process_instance(
         metadata=metadata,
         history=histories,
         metrics=metrics,
-        error=state.last_error if state and state.last_error else None,
+        error=state.get_last_error() if state and state.get_last_error() else None,
         test_result=test_result,
     )
     return output

evaluation/biocoder/run_infer.py

@@ -311,7 +311,7 @@ def process_instance(
         metadata=metadata,
         history=histories,
         metrics=metrics,
-        error=state.last_error if state and state.last_error else None,
+        error=state.get_last_error() if state and state.get_last_error() else None,
         test_result=test_result,
     )
     return output

evaluation/bird/run_infer.py

@@ -440,7 +440,7 @@ def execute_sql(db_path, sql):
         metadata=metadata,
         history=histories,
         metrics=metrics,
-        error=state.last_error if state and state.last_error else None,
+        error=state.get_last_error() if state and state.get_last_error() else None,
         test_result=test_result,
     )
     return output

evaluation/browsing_delegation/run_infer.py

@@ -121,7 +121,7 @@ def process_instance(
         metadata=metadata,
         history=histories,
         metrics=metrics,
-        error=state.last_error if state and state.last_error else None,
+        error=state.get_last_error() if state and state.get_last_error() else None,
         test_result={
             'query': instance.instruction,
             'action': last_delegate_action,

evaluation/gaia/run_infer.py

@@ -213,7 +213,7 @@ def process_instance(
         metadata=metadata,
         history=histories,
         metrics=metrics,
-        error=state.last_error if state and state.last_error else None,
+        error=state.get_last_error() if state and state.get_last_error() else None,
         test_result=test_result,
     )
     return output

evaluation/gorilla/run_infer.py

@@ -121,7 +121,7 @@ def process_instance(
         metadata=metadata,
         history=histories,
         metrics=metrics,
-        error=state.last_error if state and state.last_error else None,
+        error=state.get_last_error() if state and state.get_last_error() else None,
         test_result={
             'text': model_answer_raw,
             'correct': correct,

evaluation/gpqa/run_infer.py

@@ -302,7 +302,7 @@ def process_instance(
         metadata=metadata,
         history=state.history.compatibility_for_eval_history_pairs(),
         metrics=metrics,
-        error=state.last_error if state and state.last_error else None,
+        error=state.get_last_error() if state and state.get_last_error() else None,
         test_result={
             'result': test_result,
             'found_answers': found_answers,

evaluation/humanevalfix/run_infer.py

@@ -264,7 +264,7 @@ def process_instance(
         metadata=metadata,
         history=histories,
         metrics=metrics,
-        error=state.last_error if state and state.last_error else None,
+        error=state.get_last_error() if state and state.get_last_error() else None,
         test_result=test_result,
     )
     return output

evaluation/integration_tests/run_infer.py

@@ -134,7 +134,7 @@ def process_instance(
         metadata=metadata,
         history=histories,
         metrics=metrics,
-        error=state.last_error if state and state.last_error else None,
+        error=state.get_last_error() if state and state.get_last_error() else None,
         test_result=test_result.model_dump(),
     )
     return output

evaluation/logic_reasoning/run_infer.py

@@ -256,7 +256,7 @@ def process_instance(
         metadata=metadata,
         history=histories,
         metrics=metrics,
-        error=state.last_error if state and state.last_error else None,
+        error=state.get_last_error() if state and state.get_last_error() else None,
         test_result=test_result,
     )
     return output

evaluation/miniwob/run_infer.py

@@ -173,7 +173,7 @@ def process_instance(
         metadata=metadata,
         history=histories,
         metrics=metrics,
-        error=state.last_error if state and state.last_error else None,
+        error=state.get_last_error() if state and state.get_last_error() else None,
         test_result={
             'reward': reward,
         },

evaluation/mint/run_infer.py

@@ -212,7 +212,7 @@ def process_instance(
         metadata=metadata,
         history=histories,
         metrics=metrics,
-        error=state.last_error if state and state.last_error else None,
+        error=state.get_last_error() if state and state.get_last_error() else None,
         test_result={
             'success': task_state.success if task_state else False,
         },

evaluation/swe_bench/run_infer.py

@@ -402,10 +402,10 @@ def process_instance(
 
         # if fatal error, throw EvalError to trigger re-run
         if (
-            state.last_error
-            and 'fatal error during agent execution' in state.last_error
+            state.get_last_error()
+            and 'fatal error during agent execution' in state.get_last_error()
         ):
-            raise EvalException('Fatal error detected: ' + state.last_error)
+            raise EvalException('Fatal error detected: ' + state.get_last_error())
 
         # ======= THIS IS SWE-Bench specific =======
         # Get git patch
@@ -442,7 +442,7 @@ def process_instance(
         metadata=metadata,
         history=histories,
         metrics=metrics,
-        error=state.last_error if state and state.last_error else None,
+        error=state.get_last_error() if state and state.get_last_error() else None,
     )
     return output
 

evaluation/toolqa/run_infer.py

@@ -149,7 +149,7 @@ def process_instance(instance: Any, metadata: EvalMetadata, reset_logger: bool =
         metadata=metadata,
         history=histories,
         metrics=metrics,
-        error=state.last_error if state and state.last_error else None,
+        error=state.get_last_error() if state and state.get_last_error() else None,
     )
     return output
 

evaluation/webarena/run_infer.py

@@ -187,7 +187,7 @@ def process_instance(
         metadata=metadata,
         history=histories,
         metrics=metrics,
-        error=state.last_error if state and state.last_error else None,
+        error=state.get_last_error() if state and state.get_last_error() else None,
         test_result={
             'reward': reward,
         },

openhands/controller/agent_controller.py

@@ -133,21 +133,19 @@ async def update_state_after_step(self):
         # update metrics especially for cost. Use deepcopy to avoid it being modified by agent.reset()
         self.state.local_metrics = copy.deepcopy(self.agent.llm.metrics)
 
-    async def report_error(self, message: str, exception: Exception | None = None):
-        """Reports an error to the user and sends the exception to the LLM next step, in the hope it can self-correct.
-
-        This method should be called for a particular type of errors, which have:
-        - a user-friendly message, which will be shown in the chat box. This should not be a raw exception message.
-        - an ErrorObservation that can be sent to the LLM by the user role, with the exception message, so it can self-correct next time.
-        """
-        self.state.last_error = message
-        if exception:
-            self.state.last_error += f': {exception}'
+    async def _react_to_error(
+        self,
+        message: str,
+        exception: Exception | None = None,
+        new_state: AgentState | None = None,
+    ):
+        if new_state is not None:
+            await self.set_agent_state_to(new_state)
         detail = str(exception) if exception is not None else ''
         if exception is not None and isinstance(exception, litellm.AuthenticationError):
-            detail = 'Please check your credentials. Is your API key correct?'
-        self.event_stream.add_event(
-            ErrorObservation(f'{message}:{detail}'), EventSource.USER
+            detail += '\nPlease check your credentials. Is your API key correct?'
+        await self.event_stream.async_add_event(
+            ErrorObservation(f'{message}:{detail}'), EventSource.AGENT
         )
 
     async def start_step_loop(self):
@@ -164,10 +162,11 @@ async def start_step_loop(self):
                 traceback.print_exc()
                 logger.error(f'Error while running the agent: {e}')
                 logger.error(traceback.format_exc())
-                await self.report_error(
-                    'There was an unexpected error while running the agent', exception=e
+                await self._react_to_error(
+                    'There was an unexpected error while running the agent',
+                    exception=e,
+                    new_state=AgentState.ERROR,
                 )
-                await self.set_agent_state_to(AgentState.ERROR)
                 break
 
             await asyncio.sleep(0.1)
@@ -254,9 +253,6 @@ async def _handle_observation(self, observation: Observation):
             if self.state.agent_state == AgentState.ERROR:
                 self.state.metrics.merge(self.state.local_metrics)
         elif isinstance(observation, FatalErrorObservation):
-            await self.report_error(
-                'There was a fatal error during agent execution: ' + str(observation)
-            )
             await self.set_agent_state_to(AgentState.ERROR)
             self.state.metrics.merge(self.state.local_metrics)
 
@@ -330,7 +326,9 @@ async def set_agent_state_to(self, new_state: AgentState):
             else:
                 confirmation_state = ActionConfirmationStatus.REJECTED
             self._pending_action.confirmation_state = confirmation_state  # type: ignore[attr-defined]
-            self.event_stream.add_event(self._pending_action, EventSource.AGENT)
+            await self.event_stream.async_add_event(
+                self._pending_action, EventSource.AGENT
+            )
 
         self.state.agent_state = new_state
         self.event_stream.add_event(
@@ -443,7 +441,7 @@ async def _step(self) -> None:
         except (LLMMalformedActionError, LLMNoActionError, LLMResponseError) as e:
             # report to the user
             # and send the underlying exception to the LLM for self-correction
-            await self.report_error(str(e))
+            await self._react_to_error(str(e))
             return
 
         if action.runnable:
@@ -462,15 +460,15 @@ async def _step(self) -> None:
                 == ActionConfirmationStatus.AWAITING_CONFIRMATION
             ):
                 await self.set_agent_state_to(AgentState.AWAITING_USER_CONFIRMATION)
-            self.event_stream.add_event(action, EventSource.AGENT)
+            await self.event_stream.async_add_event(action, EventSource.AGENT)
 
         await self.update_state_after_step()
         logger.info(action, extra={'msg_type': 'ACTION'})
 
         if self._is_stuck():
-            # This need to go BEFORE report_error to sync metrics
-            await self.set_agent_state_to(AgentState.ERROR)
-            await self.report_error('Agent got stuck in a loop')
+            await self._react_to_error(
+                'Agent got stuck in a loop', new_state=AgentState.ERROR
+            )
 
     async def _delegate_step(self):
         """Executes a single step of the delegate agent."""
@@ -489,7 +487,7 @@ async def _delegate_step(self):
             self.delegate = None
             self.delegateAction = None
 
-            await self.report_error('Delegator agent encountered an error')
+            await self._react_to_error('Delegator agent encountered an error')
         elif delegate_state in (AgentState.FINISHED, AgentState.REJECTED):
             logger.info(
                 f'[Agent Controller {self.id}] Delegate agent has finished execution'
@@ -518,7 +516,7 @@ async def _delegate_step(self):
             # clean up delegate status
             self.delegate = None
             self.delegateAction = None
-            self.event_stream.add_event(obs, EventSource.AGENT)
+            await self.event_stream.async_add_event(obs, EventSource.AGENT)
         return
 
     async def _handle_traffic_control(
@@ -538,20 +536,17 @@ async def _handle_traffic_control(
         else:
             self.state.traffic_control_state = TrafficControlState.THROTTLING
             if self.headless_mode:
-                # This need to go BEFORE report_error to sync metrics
-                await self.set_agent_state_to(AgentState.ERROR)
-                # set to ERROR state if running in headless mode
-                # since user cannot resume on the web interface
-                await self.report_error(
+                await self._react_to_error(
                     f'Agent reached maximum {limit_type} in headless mode, task stopped. '
-                    f'Current {limit_type}: {current_value:.2f}, max {limit_type}: {max_value:.2f}'
+                    f'Current {limit_type}: {current_value:.2f}, max {limit_type}: {max_value:.2f}',
+                    new_state=AgentState.ERROR,
                 )
             else:
-                await self.set_agent_state_to(AgentState.PAUSED)
-                await self.report_error(
+                await self._react_to_error(
                     f'Agent reached maximum {limit_type}, task paused. '
                     f'Current {limit_type}: {current_value:.2f}, max {limit_type}: {max_value:.2f}. '
-                    f'{TRAFFIC_CONTROL_REMINDER}'
+                    f'{TRAFFIC_CONTROL_REMINDER}',
+                    new_state=AgentState.PAUSED,
                 )
             stop_step = True
         return stop_step

openhands/controller/state/state.py

@@ -11,6 +11,7 @@
     MessageAction,
 )
 from openhands.events.action.agent import AgentFinishAction
+from openhands.events.observation import ErrorObservation, FatalErrorObservation
 from openhands.llm.metrics import Metrics
 from openhands.memory.history import ShortTermHistory
 from openhands.storage.files import FileStore
@@ -80,7 +81,6 @@ class State:
     history: ShortTermHistory = field(default_factory=ShortTermHistory)
     inputs: dict = field(default_factory=dict)
     outputs: dict = field(default_factory=dict)
-    last_error: str | None = None
     agent_state: AgentState = AgentState.LOADING
     resume_state: AgentState | None = None
     traffic_control_state: TrafficControlState = TrafficControlState.NORMAL
@@ -124,9 +124,6 @@ def restore_from_session(sid: str, file_store: FileStore) -> 'State':
         else:
             state.resume_state = None
 
-        # don't carry last_error anymore after restore
-        state.last_error = None
-
         # first state after restore
         state.agent_state = AgentState.LOADING
         return state
@@ -157,6 +154,14 @@ def __setstate__(self, state):
 
         # remove the restored data from the state if any
 
+    def get_last_error(self) -> str:
+        for event in self.history.get_events(reverse=True):
+            if isinstance(event, ErrorObservation) or isinstance(
+                event, FatalErrorObservation
+            ):
+                return event.content
+        return ''
+
     def get_current_user_intent(self):
         """Returns the latest user message and image(if provided) that appears after a FinishAction, or the first (the task) if nothing was finished yet."""
         last_user_message = None

openhands/events/stream.py

@@ -138,6 +138,10 @@ def add_event(self, event: Event, source: EventSource):
             asyncio.run(self.async_add_event(event, source))
 
     async def async_add_event(self, event: Event, source: EventSource):
+        if hasattr(event, '_id') and event.id is not None:
+            raise ValueError(
+                'Event already has an ID. It was probably added back to the EventStream from inside a handler, trigging a loop.'
+            )
         with self._lock:
             event._id = self._cur_id  # type: ignore [attr-defined]
             self._cur_id += 1