Sprint 1: EventBus — async pub/sub with backpressure, replay, lane isolation

purpose_agent/runtime/event_bus.py

@@ -0,0 +1,236 @@
+"""
+event_bus.py — Async pub/sub event bus with backpressure, replay, and lane isolation.
+
+Features:
+  - Multiple subscribers (async generators)
+  - Bounded queues with backpressure (slow consumer doesn't kill producer)
+  - Terminal events (run.finished, run.error) are NEVER dropped
+  - Per-lane sequence tracking for parallel execution
+  - Event replay from history for late subscribers
+  - Thread-safe for mixed sync/async usage
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import threading
+import time
+from collections import defaultdict
+from typing import Any, AsyncIterator, Callable
+
+from purpose_agent.runtime.events import PAEvent, EventKind, Visibility, TERMINAL_KINDS
+
+logger = logging.getLogger(__name__)
+
+
+class EventBus:
+    """
+    Async event bus for the Purpose Agent runtime.
+
+    Usage:
+        bus = EventBus(max_queue_size=1000)
+
+        # Subscribe
+        async for event in bus.subscribe(visibility=Visibility.PUBLIC):
+            print(event.kind, event.payload)
+
+        # Publish (can be called from sync or async context)
+        bus.emit(event)
+
+        # Replay (for late subscribers or crash recovery)
+        events = bus.replay(run_id="abc123")
+    """
+
+    def __init__(self, max_queue_size: int = 1000, max_history: int = 10000):
+        self._subscribers: list[asyncio.Queue] = []
+        self._subscriber_filters: list[dict[str, Any]] = []
+        self._max_queue_size = max_queue_size
+        self._history: list[PAEvent] = []
+        self._max_history = max_history
+        self._lane_seqs: dict[str, int] = defaultdict(int)
+        self._lock = threading.Lock()
+        self._closed = False
+
+    def emit(self, event: PAEvent) -> None:
+        """
+        Publish an event to all subscribers.
+
+        Terminal events are force-delivered even if queue is full.
+        Non-terminal events are dropped if subscriber is too slow (backpressure).
+        """
+        if self._closed:
+            return
+
+        # Safety check: reject events with raw chain-of-thought
+        if event.has_hidden_cot():
+            logger.warning(
+                f"EventBus: REJECTED event {event.span_id} — contains hidden chain-of-thought. "
+                f"Use EventKind.REASONING_SUMMARY with safe payload instead."
+            )
+            return
+
+        with self._lock:
+            # Track history for replay
+            if len(self._history) >= self._max_history:
+                self._history = self._history[-self._max_history // 2:]
+            self._history.append(event)
+
+        # Deliver to subscribers
+        for i, queue in enumerate(self._subscribers):
+            filters = self._subscriber_filters[i]
+            if not self._matches_filter(event, filters):
+                continue
+
+            if event.is_terminal:
+                # Terminal events are NEVER dropped — block until delivered
+                try:
+                    queue.put_nowait(event)
+                except asyncio.QueueFull:
+                    # Force: remove oldest non-terminal to make room
+                    try:
+                        queue.get_nowait()
+                        queue.put_nowait(event)
+                    except (asyncio.QueueEmpty, asyncio.QueueFull):
+                        pass
+            else:
+                # Non-terminal: drop if full (backpressure)
+                try:
+                    queue.put_nowait(event)
+                except asyncio.QueueFull:
+                    logger.debug(f"EventBus: backpressure — dropped {event.kind.value} for subscriber {i}")
+
+    def emit_sync(self, event: PAEvent) -> None:
+        """Alias for emit() — usable from synchronous code."""
+        self.emit(event)
+
+    async def subscribe(
+        self,
+        visibility: Visibility | None = None,
+        lane_id: str | None = None,
+        kinds: set[EventKind] | None = None,
+        run_id: str | None = None,
+    ) -> AsyncIterator[PAEvent]:
+        """
+        Subscribe to events matching the given filters.
+
+        Yields PAEvents as they arrive. Stops when bus is closed.
+        """
+        queue: asyncio.Queue = asyncio.Queue(maxsize=self._max_queue_size)
+        filters = {
+            "visibility": visibility,
+            "lane_id": lane_id,
+            "kinds": kinds,
+            "run_id": run_id,
+        }
+
+        self._subscribers.append(queue)
+        self._subscriber_filters.append(filters)
+        idx = len(self._subscribers) - 1
+
+        try:
+            while not self._closed:
+                try:
+                    event = await asyncio.wait_for(queue.get(), timeout=0.5)
+                    yield event
+                    if event.kind in (EventKind.RUN_FINISHED, EventKind.RUN_ERROR):
+                        break
+                except asyncio.TimeoutError:
+                    continue
+        finally:
+            # Cleanup
+            if idx < len(self._subscribers):
+                self._subscribers[idx] = asyncio.Queue()  # Orphan the queue
+
+    def replay(
+        self,
+        run_id: str | None = None,
+        lane_id: str | None = None,
+        since_seq: int = 0,
+    ) -> list[PAEvent]:
+        """
+        Replay events from history.
+
+        Useful for:
+          - Late subscribers catching up
+          - Crash recovery
+          - Debugging
+        """
+        with self._lock:
+            events = list(self._history)
+
+        if run_id:
+            events = [e for e in events if e.run_id == run_id]
+        if lane_id:
+            events = [e for e in events if e.lane_id == lane_id]
+        if since_seq > 0:
+            events = [e for e in events if e.seq > since_seq]
+
+        return events
+
+    def next_seq(self, lane_id: str = "main") -> int:
+        """Get and increment the next sequence number for a lane."""
+        with self._lock:
+            self._lane_seqs[lane_id] += 1
+            return self._lane_seqs[lane_id]
+
+    def close(self) -> None:
+        """Close the bus — no more events will be accepted."""
+        self._closed = True
+
+    @property
+    def history_size(self) -> int:
+        return len(self._history)
+
+    @property
+    def subscriber_count(self) -> int:
+        return len(self._subscribers)
+
+    @staticmethod
+    def _matches_filter(event: PAEvent, filters: dict[str, Any]) -> bool:
+        """Check if an event matches subscriber filters."""
+        if filters.get("visibility") and event.visibility != filters["visibility"]:
+            # Allow public events to pass through internal/debug filters
+            if event.visibility.value > filters["visibility"].value:
+                return False
+        if filters.get("lane_id") and event.lane_id != filters["lane_id"]:
+            return False
+        if filters.get("kinds") and event.kind not in filters["kinds"]:
+            return False
+        if filters.get("run_id") and event.run_id != filters["run_id"]:
+            return False
+        return True
+
+
+def parallel_merge(lane_events: dict[str, list[PAEvent]]) -> list[PAEvent]:
+    """
+    Merge events from multiple parallel lanes into a single deterministic stream.
+
+    Ordering: events are interleaved by timestamp, with lane_id as tiebreaker.
+    Each lane's sequence numbers remain monotonic.
+
+    Usage:
+        merged = parallel_merge({
+            "lane_a": [event1, event2],
+            "lane_b": [event3, event4],
+        })
+    """
+    all_events = []
+    for lane_id, events in lane_events.items():
+        for event in events:
+            all_events.append(event)
+
+    # Sort by timestamp, then lane_id for determinism
+    all_events.sort(key=lambda e: (e.ts, e.lane_id, e.seq))
+
+    # Verify lane-local monotonicity
+    lane_last_seq: dict[str, int] = {}
+    for event in all_events:
+        last = lane_last_seq.get(event.lane_id, -1)
+        if event.seq < last:
+            logger.warning(
+                f"parallel_merge: non-monotonic seq in lane {event.lane_id}: "
+                f"got {event.seq} after {last}"
+            )
+        lane_last_seq[event.lane_id] = event.seq
+
+    return all_events