Optimize EventManager waiting logic and error handling

Author: yarikdevcom

agents-core/vision_agents/core/events/manager.py

@@ -142,6 +142,7 @@ def __init__(self, ignore_unknown_events: bool = True):
         self._shutdown = False
         self._silent_events: set[type] = set()
         self._handler_tasks: Dict[uuid.UUID, asyncio.Task[Any]] = {}
+        self._received_event = asyncio.Event()
 
         self.register(ExceptionEvent)
         self.register(HealthCheckEvent)
@@ -213,6 +214,7 @@ def merge(self, em: "EventManager"):
         em._queue = self._queue
         em._silent_events = self._silent_events
         em._processing_task = None  # Clear the stopped task reference
+        em._received_event = self._received_event
 
     def register_events_from_module(
         self, module, prefix="", ignore_not_compatible=True
@@ -467,6 +469,8 @@ def send(self, *events):
             if event:
                 self._queue.append(event)
 
+        self._received_event.set()
+
     async def wait(self, timeout: float = 10.0):
         """
         Wait for all queued events to be processed.
@@ -486,7 +490,7 @@ async def wait(self, timeout: float = 10.0):
     def _start_processing_task(self):
         """Start the background event processing task."""
         if self._processing_task and not self._processing_task.done():
-            return  # Already running
+            return
 
         loop = asyncio.get_running_loop()
         self._processing_task = loop.create_task(self._process_events_loop())
@@ -521,13 +525,15 @@ async def _process_events_loop(self):
                 )
                 for task_id in cleanup_ids:
                     self._handler_tasks.pop(task_id)
-                await asyncio.sleep(0.0001)
+
+                await self._received_event.wait()
+                self._received_event.clear()
 
     async def _run_handler(self, handler, event):
         try:
             return await handler(event)
         except Exception as exc:
-            self._queue.appendleft(ExceptionEvent(exc, handler))  # type: ignore[arg-type]
+            self.send(ExceptionEvent(exc, handler))
             module_name = getattr(handler, "__module__", "unknown")
             logger.exception(
                 f"Error calling handler {handler.__name__} from {module_name} for event {event.type}"
@@ -536,10 +542,9 @@ async def _run_handler(self, handler, event):
     async def _process_single_event(self, event):
         """Process a single event."""
         for handler in self._handlers.get(event.type, []):
-            #module_name = getattr(handler, '__module__', 'unknown')
+            module_name = getattr(handler, '__module__', 'unknown')
             if event.type not in self._silent_events:
-                pass
-                #logger.info(f"Called handler {handler.__name__} from {module_name} for event {event.type}")
+                logger.debug(f"Called handler {handler.__name__} from {module_name} for event {event.type}")
 
             loop = asyncio.get_running_loop()
             handler_task = loop.create_task(self._run_handler(handler, event))