# backend/easter_eggs/engine_integration.py # PURPOSE: Intercepts user inputs or frontend signals to trigger Easter Eggs # before they hit the LLM. If an Easter Egg triggers, it bypasses the LLM # and returns the handcrafted response immediately. import re import logging from .registry import EASTER_EGG_REGISTRY, EasterEgg # Fuzzy matcher for trigger phrases def _is_match(user_input: str, egg: EasterEgg) -> bool: if not user_input or not egg.trigger_phrases: return False cleaned_input = re.sub(r'[^\w\s]', '', user_input.lower()).strip() for trigger in egg.trigger_phrases: cleaned_trigger = re.sub(r'[^\w\s]', '', trigger.lower()).strip() if cleaned_trigger == cleaned_input: return True # Allow slight padding (e.g. "hey jarvis i am iron man") if cleaned_trigger in cleaned_input and len(cleaned_input) < len(cleaned_trigger) + 15: return True return False async def check_for_easter_egg(user_text: str = None, ui_event: str = None, context: dict = None) -> dict: """ Checks if the current input triggers any registered Easter Egg. Returns the resolved response dict, or None if no egg triggered. """ if context is None: context = {} for egg in EASTER_EGG_REGISTRY: triggered = False # 1. Check UI/Event triggers if ui_event and egg.ui_trigger == ui_event: triggered = True # 2. Check text phrases if user_text and not triggered: if _is_match(user_text, egg): triggered = True # 3. Check proactive (time-based) triggers if not user_text and not ui_event and hasattr(egg, "proactive") and egg.proactive: # The handler itself decides if it should trigger triggered = True if triggered: try: logging.getLogger(__name__).info(f"[EASTER EGG] Triggered: {egg.id}") response = await egg.handler(context) if response: # Inject metadata so the frontend knows how to handle it response["_is_easter_egg"] = True response["_egg_id"] = egg.id return response except Exception as e: logging.getLogger(__name__).error(f"[EASTER EGG] Failed executing {egg.id}: {e}") return None