""" Background-Loop fuer Triggers. Laeuft alle TICK_SEC Sekunden in einem asyncio Task, geht ueber alle active Triggers und entscheidet ob sie feuern muessen. Feuern bedeutet: 1. Trigger-Manifest update (fire_count++, last_fired_at, ggf. deaktivieren) 2. Log-Eintrag schreiben 3. agent.chat() mit einem system-Praefix aufrufen (NICHT als 'user'!) → ARIA bekommt das wie eine Push-Nachricht und kann antworten """ from __future__ import annotations import asyncio import logging from datetime import datetime, timezone from typing import Optional import triggers as triggers_mod import watcher as watcher_mod logger = logging.getLogger(__name__) TICK_SEC = 30 def _now_iso() -> str: return datetime.now(timezone.utc).isoformat() def _parse_iso(s: str) -> Optional[datetime]: if not s: return None try: return datetime.fromisoformat(s.replace("Z", "+00:00")) except Exception: return None def _should_fire(trigger: dict, vars_: dict, now: datetime) -> bool: if not trigger.get("active", True): return False t = trigger.get("type", "") if t == "timer": fires_at = _parse_iso(trigger.get("fires_at", "")) if not fires_at: return False if fires_at.tzinfo is None: fires_at = fires_at.replace(tzinfo=timezone.utc) return now >= fires_at if t == "watcher": # Check-Interval respektieren (sonst pollen wir zu hektisch) check_interval = int(trigger.get("check_interval_sec", 300)) last_checked = _parse_iso(trigger.get("last_checked_at", "")) if last_checked: if last_checked.tzinfo is None: last_checked = last_checked.replace(tzinfo=timezone.utc) if (now - last_checked).total_seconds() < check_interval: return False # Throttle: erst feuern wenn last_fired lange genug her ist last_fired = _parse_iso(trigger.get("last_fired_at", "")) throttle = int(trigger.get("throttle_sec", 3600)) if last_fired: if last_fired.tzinfo is None: last_fired = last_fired.replace(tzinfo=timezone.utc) if (now - last_fired).total_seconds() < throttle: return False # Condition pruefen cond = (trigger.get("condition") or "").strip() if not cond: return False try: return watcher_mod.evaluate(cond, vars_) except Exception as e: logger.warning("Trigger %s: Condition '%s' fehlerhaft: %s", trigger.get("name"), cond, e) return False if t == "cron": # TODO: später, wenn jemand Bock auf Cron-Parser hat return False return False async def _fire(trigger: dict, agent_factory) -> None: """Ruft ARIA mit einer System-Praefix-Nachricht auf.""" name = trigger.get("name", "?") message = trigger.get("message") or "(ohne Nachricht)" ttype = trigger.get("type", "?") # Manifest updaten try: triggers_mod.mark_fired(name) except Exception as e: logger.warning("mark_fired %s: %s", name, e) # Log triggers_mod.append_log(name, {"event": "fired", "type": ttype, "message": message}) # System-Nachricht an ARIA: nicht als User, sondern als Hinweis prompt = ( f"[Trigger ausgelöst: '{name}', Typ: {ttype}] " f"Geplante Nachricht: \"{message}\". " f"Sage Stefan jetzt diese Information, in deinem Stil. " f"Wenn der Trigger ein Watcher war (Bedingung wurde erfuellt), " f"erwaehne kurz worum es geht. Antworte direkt, keine Rueckfrage." ) try: agent = agent_factory() reply = agent.chat(prompt, source="trigger") logger.info("[trigger] %s gefeuert → ARIA-Reply: %s", name, reply[:80]) triggers_mod.append_log(name, {"event": "reply", "text": reply[:500]}) except Exception as e: logger.exception("Trigger %s feuern fehlgeschlagen: %s", name, e) triggers_mod.append_log(name, {"event": "error", "error": str(e)[:300]}) async def _tick(agent_factory) -> None: """Ein Pruefdurchlauf. Geht ueber alle Triggers, feuert was zu feuern ist.""" try: all_triggers = triggers_mod.list_triggers(active_only=True) except Exception as e: logger.warning("triggers.list: %s", e) return if not all_triggers: return now = datetime.now(timezone.utc) # Variablen einmal pro Tick sammeln (nicht pro Trigger — Disk-Stat ist teuer) try: vars_ = watcher_mod.collect_variables() except Exception as e: logger.warning("collect_variables: %s", e) vars_ = {} # Watcher: last_checked_at jetzt updaten (auch wenn nicht gefeuert wird, # damit der Check-Interval respektiert wird) for t in all_triggers: if t.get("type") == "watcher": try: t["last_checked_at"] = _now_iso() triggers_mod.write(t["name"], t) except Exception: pass for trigger in all_triggers: try: if _should_fire(trigger, vars_, now): # Feuern als eigener Task — wenn ARIA langsam antwortet, # darf der naechste Tick nicht blockieren asyncio.create_task(_fire(trigger, agent_factory)) except Exception as e: logger.warning("Trigger-Check %s: %s", trigger.get("name"), e) async def run_loop(agent_factory) -> None: """Endlosschleife — wird vom main lifespan gestartet + gestoppt.""" logger.info("Trigger-Loop gestartet (TICK_SEC=%d)", TICK_SEC) while True: try: await _tick(agent_factory) except Exception as e: logger.exception("Tick-Fehler: %s", e) await asyncio.sleep(TICK_SEC)