src/common/events.py

"""Typed Runtime Event Bus for FlyBrain V8/V9.

Implements Section 84:
Publishes and dispatches typed simulation, cognitive, asset, and resource events.
Thread-safe, non-blocking listener invocation with bounded history.
"""
import time
import logging
import threading
from enum import Enum
from dataclasses import dataclass, field, asdict
from typing import Dict, Any, List, Callable, Optional

logger = logging.getLogger("FlyBrain.EventBus")


class EventType(str, Enum):
    # Simulation & Physics
    WORLD_TICK = "WORLD_TICK"
    CHUNK_LOADED = "CHUNK_LOADED"
    CHUNK_UNLOADED = "CHUNK_UNLOADED"
    ORGANISM_MOVED = "ORGANISM_MOVED"
    COLLISION = "COLLISION"
    CONTACT = "CONTACT"
    
    # Cognition & Biology
    NEURON_SPIKE = "NEURON_SPIKE"
    GOAL_CHANGED = "GOAL_CHANGED"
    MEMORY_UPDATED = "MEMORY_UPDATED"
    SOCIAL_INTERACTION = "SOCIAL_INTERACTION"
    DREAM_STARTED = "DREAM_STARTED"
    DREAM_COMPLETED = "DREAM_COMPLETED"
    
    # Generative AI & Assets
    IMAGE_GENERATION_STARTED = "IMAGE_GENERATION_STARTED"
    IMAGE_GENERATED = "IMAGE_GENERATED"
    ASSET_GENERATION_STARTED = "ASSET_GENERATION_STARTED"
    ASSET_GENERATED = "ASSET_GENERATED"
    ASSET_VALIDATED = "ASSET_VALIDATED"
    RIG_CREATED = "RIG_CREATED"
    MOTION_STARTED = "MOTION_STARTED"
    MOTION_COMPLETED = "MOTION_COMPLETED"
    
    # Model & Resource Management
    MODEL_LOADING = "MODEL_LOADING"
    MODEL_LOADED = "MODEL_LOADED"
    MODEL_OFFLOADED = "MODEL_OFFLOADED"
    MODEL_FAILED = "MODEL_FAILED"
    RESOURCE_PRESSURE_CHANGED = "RESOURCE_PRESSURE_CHANGED"
    
    # Persistence & Resilience
    BACKUP_CREATED = "BACKUP_CREATED"
    BACKUP_VERIFIED = "BACKUP_VERIFIED"
    BACKUP_RESTORED = "BACKUP_RESTORED"
    WATCHDOG_RECOVERY = "WATCHDOG_RECOVERY"


@dataclass
class Event:
    type: EventType
    timestamp: float = field(default_factory=time.time)
    source: str = "core"
    payload: Dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> Dict[str, Any]:
        return {
            "type": self.type.value,
            "timestamp": self.timestamp,
            "source": self.source,
            "payload": self.payload
        }


class EventBus:
    _instance = None
    _lock = threading.RLock()

    def __new__(cls):
        with cls._lock:
            if cls._instance is None:
                cls._instance = super(EventBus, cls).__new__(cls)
                cls._instance._initialized = False
            return cls._instance

    def __init__(self):
        if getattr(self, "_initialized", False):
            return
        self._lock = threading.RLock()
        self._subscribers: Dict[EventType, List[Callable[[Event], None]]] = {
            et: [] for et in EventType
        }
        self._all_subscribers: List[Callable[[Event], None]] = []
        self._history: List[Event] = []
        self._max_history = 1000
        self._initialized = True

    def subscribe(self, event_type: EventType, callback: Callable[[Event], None]) -> None:
        """Subscribes a callback to a specific typed event."""
        with self._lock:
            if event_type in self._subscribers:
                self._subscribers[event_type].append(callback)

    def subscribe_all(self, callback: Callable[[Event], None]) -> None:
        """Subscribes a callback to all events (e.g. for WebSocket telemetry)."""
        with self._lock:
            self._all_subscribers.append(callback)

    def unsubscribe(self, event_type: EventType, callback: Callable[[Event], None]) -> None:
        with self._lock:
            if event_type in self._subscribers and callback in self._subscribers[event_type]:
                self._subscribers[event_type].remove(callback)

    def publish(self, event: Event) -> None:
        """Publishes an event to all subscribers and records to bounded history."""
        with self._lock:
            self._history.append(event)
            if len(self._history) > self._max_history:
                self._history.pop(0)

            # Copy listener lists for safe iteration outside lock
            listeners = list(self._subscribers.get(event.type, []))
            wildcard = list(self._all_subscribers)

        for cb in listeners:
            try:
                cb(event)
            except Exception as e:
                logger.error(f"Error in listener for {event.type.value}: {e}")

        for cb in wildcard:
            try:
                cb(event)
            except Exception as e:
                logger.error(f"Error in wildcard listener for {event.type.value}: {e}")

    def get_recent(self, count: int = 50, event_type: Optional[EventType] = None) -> List[Dict[str, Any]]:
        """Returns recent events in chronological order."""
        with self._lock:
            if event_type is not None:
                filtered = [e for e in self._history if e.type == event_type]
            else:
                filtered = self._history
            return [e.to_dict() for e in filtered[-count:]]


def get_event_bus() -> EventBus:
    return EventBus()


# ---- Backward Compatibility Biological Event Stream (V4/V5/V7) ----
import hashlib
import os

VALID_EVENTS = {
    "NEURON_BORN", "NEURON_DIFFERENTIATED", "NEURON_MOVED",
    "AXON_GROWN", "DENDRITE_GROWN", "SYNAPSE_CREATED",
    "SYNAPSE_STRENGTHENED", "SYNAPSE_WEAKENED", "SYNAPSE_PRUNED",
    "NEURON_DIED", "MEMORY_CREATED", "MEMORY_RECALLED",
    "MEMORY_CONSOLIDATED", "SLEEP_STARTED", "DREAM_STARTED",
    "DREAM_ENDED", "TEACHING_STARTED", "TEACHING_ENDED",
    "LEARNING_EVENT", "REPRODUCTION", "MUTATION", "CROSSOVER",
    "ORGANISM_BORN", "ORGANISM_DIED", "GENERATION_STARTED",
    "GENERATION_ENDED", "WORLD_STEP", "SNAPSHOT_SAVED",
}


class EventLog:
    def __init__(self):
        self.events: List[Dict[str, Any]] = []
        self._tick_counter = 0

    def log(self, event_type: str, tick: int, organism_id: str = "",
            generation: int = 0, payload: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        if event_type not in VALID_EVENTS:
            raise ValueError(f"Unknown event type: {event_type}")
        ev = {
            "seq": len(self.events),
            "type": event_type,
            "tick": int(tick),
            "organism_id": organism_id,
            "generation": int(generation),
            "payload": payload or {},
        }
        self.events.append(ev)
        return ev

    def __len__(self):
        return len(self.events)

    def filter(self, event_type: str = "", organism_id: str = "") -> List[Dict[str, Any]]:
        out = self.events
        if event_type:
            out = [e for e in out if e["type"] == event_type]
        if organism_id:
            out = [e for e in out if e["organism_id"] == organism_id]
        return out

    def compute_hash(self) -> str:
        import json as _json
        h = hashlib.sha256()
        for e in self.events:
            h.update(_json.dumps(e, sort_keys=True).encode())
        return h.hexdigest()

    def to_jsonl(self, path: str):
        import json as _json
        os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True)
        with open(path, "w", encoding="utf-8") as f:
            for e in self.events:
                f.write(_json.dumps(e, sort_keys=True) + "\n")

    @classmethod
    def from_jsonl(cls, path: str) -> "EventLog":
        import json as _json
        log = cls()
        with open(path, "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if line:
                    log.events.append(_json.loads(line))
        return log