auto-claude: subtask-2-1 - Create observer/event_bus.py with SessionEventBus

Co-Authored-By: Claude Opus 4.6 <[email protected]>

apps/backend/observer/event_bus.py

@@ -0,0 +1,114 @@
+"""
+Session event bus for the observer memory system.
+
+Provides an asyncio-based event bus for routing agent session events
+to the observer system. Uses a bounded queue to prevent memory issues
+and includes source tagging to prevent observer feedback loops.
+"""
+
+import asyncio
+import logging
+from collections.abc import AsyncIterator
+from dataclasses import dataclass
+from typing import Any
+
+from .models import SessionEvent
+
+logger = logging.getLogger(__name__)
+
+OBSERVER_SOURCE = "observer"
+
+
+@dataclass
+class EventBusStats:
+    """Tracks event bus throughput statistics."""
+
+    events_published: int = 0
+    events_dropped: int = 0
+
+
+class SessionEventBus:
+    """Async event bus for routing agent session events.
+
+    Wraps an asyncio.Queue with non-blocking publish semantics,
+    observer-loop prevention via source tagging, and basic stats tracking.
+
+    Args:
+        maxsize: Maximum number of buffered events. Defaults to 10000.
+    """
+
+    def __init__(self, maxsize: int = 10000) -> None:
+        self._queue: asyncio.Queue[SessionEvent] = asyncio.Queue(maxsize=maxsize)
+        self._stats = EventBusStats()
+
+    async def publish(self, event: dict[str, Any] | SessionEvent) -> None:
+        """Publish an event to the bus (non-blocking).
+
+        Events with source='observer' are silently ignored to prevent
+        feedback loops. If the queue is full, the event is dropped
+        and a debug message is logged.
+
+        Args:
+            event: A SessionEvent instance or a dict with event_type, data, source.
+        """
+        if isinstance(event, dict):
+            event = SessionEvent.from_dict(event)
+
+        if event.source == OBSERVER_SOURCE:
+            return
+
+        try:
+            self._queue.put_nowait(event)
+            self._stats.events_published += 1
+        except asyncio.QueueFull:
+            self._stats.events_dropped += 1
+            logger.debug(
+                "Event bus full, dropping event: type=%s source=%s",
+                event.event_type,
+                event.source,
+            )
+
+    async def subscribe(self) -> AsyncIterator[SessionEvent]:
+        """Yield events from the bus as they arrive.
+
+        Returns an async iterator that blocks waiting for the next event.
+        Intended for long-running consumer tasks.
+        """
+        while True:
+            event = await self._queue.get()
+            yield event
+
+    async def drain(self) -> list[SessionEvent]:
+        """Flush all currently buffered events and return them.
+
+        Returns:
+            List of all events that were in the queue.
+        """
+        events: list[SessionEvent] = []
+        while not self._queue.empty():
+            try:
+                events.append(self._queue.get_nowait())
+            except asyncio.QueueEmpty:
+                break
+        return events
+
+    def get_buffered_events(self) -> list[SessionEvent]:
+        """Peek at all currently buffered events without removing them.
+
+        Note: This accesses the internal queue deque directly for
+        peek semantics. The returned list is a snapshot.
+
+        Returns:
+            List of events currently in the queue.
+        """
+        # asyncio.Queue stores items in _queue (a collections.deque)
+        return list(self._queue._queue)  # type: ignore[attr-defined]
+
+    def qsize(self) -> int:
+        """Return the number of events currently in the queue."""
+        return self._queue.qsize()
+
+    @property
+    def stats(self) -> EventBusStats:
+        """Return current event bus statistics."""
+        return self._stats