meeting-room/meeting_room/events.py

"""EventEmitter and event types for the Meeting Room discussion engine.

Synchronous event bus with AsyncEventBridge for forwarding events
to asyncio consumers (WebSocket / SSE).
"""

from __future__ import annotations

import asyncio
import logging
import threading
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Callable

logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Event type constants
# ---------------------------------------------------------------------------

DISCUSSION_START = "discussion_start"
ROUND_START = "round_start"
AGENT_TURN_START = "agent_turn_start"
AGENT_MESSAGE = "agent_message"
TOOL_CALL = "tool_call"
TOOL_RESULT = "tool_result"
AGENT_ERROR = "agent_error"
ROUND_END = "round_end"
DISCUSSION_END = "discussion_end"
BOSS_TURN = "boss_turn"

# ---------------------------------------------------------------------------
# Event dataclass
# ---------------------------------------------------------------------------


@dataclass
class Event:
    """A single event emitted by the discussion engine."""

    type: str
    data: dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


# ---------------------------------------------------------------------------
# EventEmitter
# ---------------------------------------------------------------------------

# Type alias for handler callbacks
Handler = Callable[[Event], None]


class EventEmitter:
    """Synchronous pub/sub event bus.

    Usage::

        bus = EventEmitter()

        def on_message(event: Event):
            print(event.data)

        bus.on(AGENT_MESSAGE, on_message)
        bus.emit(AGENT_MESSAGE, text="Hello", agent="Alice")
        bus.off(AGENT_MESSAGE, on_message)
    """

    def __init__(self) -> None:
        self._handlers: dict[str, list[Handler]] = {}

    # -- subscription --------------------------------------------------------

    def on(self, event_type: str, handler: Handler) -> Handler:
        """Subscribe *handler* to *event_type*.

        Returns the handler so callers can use ``off`` with the same
        reference, or chain calls.
        """
        self._handlers.setdefault(event_type, []).append(handler)
        return handler

    def off(self, event_type: str, handler: Handler) -> None:
        """Unsubscribe *handler* from *event_type*.

        Silently ignores unknown event types or missing handlers.
        """
        handlers = self._handlers.get(event_type)
        if handlers is None:
            return
        try:
            handlers.remove(handler)
        except ValueError:
            pass
        # Clean up empty lists to avoid unbounded growth
        if not handlers:
            del self._handlers[event_type]

    # -- emission ------------------------------------------------------------

    def emit(self, event_type: str, **data: Any) -> Event:
        """Create an Event and call every subscribed handler synchronously.

        Handlers that raise exceptions are caught and logged; the emitter
        never crashes due to a bad handler.

        Returns the Event that was emitted (useful for testing).
        """
        event = Event(type=event_type, data=data)
        handlers = self._handlers.get(event_type, [])
        for handler in handlers:
            try:
                handler(event)
            except Exception:
                logger.exception(
                    "Handler %r raised for event %s",
                    handler,
                    event_type,
                )
        return event

    # -- introspection -------------------------------------------------------

    def handlers(self, event_type: str) -> list[Handler]:
        """Return a *copy* of the handler list for *event_type*."""
        return list(self._handlers.get(event_type, []))

    def clear(self) -> None:
        """Remove all subscriptions."""
        self._handlers.clear()


# ---------------------------------------------------------------------------
# AsyncEventBridge
# ---------------------------------------------------------------------------

# All event types the bridge forwards.
_ALL_EVENT_TYPES = [
    DISCUSSION_START,
    ROUND_START,
    AGENT_TURN_START,
    AGENT_MESSAGE,
    TOOL_CALL,
    TOOL_RESULT,
    AGENT_ERROR,
    ROUND_END,
    DISCUSSION_END,
    BOSS_TURN,
]


class AsyncEventBridge:
    """Bridges synchronous EventEmitter events to asyncio consumers.

    Subscribes to all event types on a sync ``EventEmitter`` and forwards
    each event into ``asyncio.Queue`` objects — one per consumer. Designed
    for running ``DiscussionEngine`` in a worker thread via
    ``asyncio.to_thread()`` while WebSocket / SSE handlers await events
    on the async side.

    Usage::

        bus = EventEmitter()
        bridge = AsyncEventBridge(bus)
        bridge.start(asyncio.get_event_loop())

        # In an async handler:
        queue = bridge.subscribe()
        try:
            while True:
                event = await queue.get()
                await websocket.send_json(event)
        finally:
            bridge.unsubscribe(queue)

        # Run engine in a thread:
        await asyncio.to_thread(engine.run, scenario)

        # Cleanup:
        bridge.stop()
    """

    def __init__(self, event_bus: EventEmitter) -> None:
        self._bus = event_bus
        self._loop: asyncio.AbstractEventLoop | None = None
        self._queues: list[asyncio.Queue[dict]] = []
        self._lock = threading.Lock()
        self._handler = self._on_sync_event

    # -- lifecycle -----------------------------------------------------------

    def start(self, loop: asyncio.AbstractEventLoop) -> None:
        """Subscribe to all event types on the bus."""
        self._loop = loop
        for event_type in _ALL_EVENT_TYPES:
            self._bus.on(event_type, self._handler)

    def stop(self) -> None:
        """Unsubscribe from all event types."""
        for event_type in _ALL_EVENT_TYPES:
            self._bus.off(event_type, self._handler)

    # -- consumer management ------------------------------------------------

    def subscribe(self) -> asyncio.Queue[dict]:
        """Create a consumer queue that receives all events.

        Call from async context. The caller owns the queue and must
        call ``unsubscribe()`` when done.
        """
        q: asyncio.Queue[dict] = asyncio.Queue(maxsize=1000)
        with self._lock:
            self._queues.append(q)
        return q

    def unsubscribe(self, q: asyncio.Queue[dict]) -> None:
        """Remove a consumer queue. Safe to call multiple times."""
        with self._lock:
            try:
                self._queues.remove(q)
            except ValueError:
                pass

    # -- internal ------------------------------------------------------------

    def _on_sync_event(self, event: Event) -> None:
        """Sync handler called by EventEmitter on any thread."""
        payload = {
            "type": event.type,
            "data": event.data,
            "timestamp": event.timestamp.isoformat(),
        }
        if self._loop is None or not self._loop.is_running():
            return
        with self._lock:
            queues = list(self._queues)
        for q in queues:
            self._loop.call_soon_threadsafe(self._safe_put, q, payload)

    @staticmethod
    def _safe_put(q: asyncio.Queue[dict], payload: dict) -> None:
        """Put payload into queue, dropping if full."""
        try:
            q.put_nowait(payload)
        except asyncio.QueueFull:
            logger.warning("AsyncEventBridge: dropping event, queue full")