import asyncio import threading from collections.abc import Awaitable, Callable from concurrent.futures import TimeoutError as FutureTimeoutError from typing import Any from bot.autonomy_logic import process_autonomy_goal from bot.control_plane import RISK_AUTO_CHAT, RISK_SUGGEST_STREAMER, control_plane from bot.observability import observability from bot.runtime_config import logger from bot.sentiment_engine import sentiment_engine AutoChatDispatcher = Callable[[str], Awaitable[None]] class AutonomyRuntime: def __init__(self) -> None: self._lock = threading.Lock() self._loop: asyncio.AbstractEventLoop | None = None self._loop_task: asyncio.Task[Any] | None = None self._tick_lock: asyncio.Lock | None = None self._auto_chat_dispatcher: AutoChatDispatcher | None = None self._mode = "eventsub" def bind( self, *, loop: asyncio.AbstractEventLoop, mode: str, auto_chat_dispatcher: AutoChatDispatcher | None = None, ) -> None: with self._lock: self._loop = loop self._mode = (mode or "eventsub").strip().lower() or "eventsub" self._auto_chat_dispatcher = auto_chat_dispatcher loop.call_soon_threadsafe(self._ensure_loop_task) def unbind(self) -> None: with self._lock: task = self._loop_task self._loop = None self._loop_task = None self._tick_lock = None self._auto_chat_dispatcher = None self._mode = "eventsub" if task is not None: task.cancel() control_plane.set_loop_running(False) def _ensure_loop_task(self) -> None: with self._lock: loop = self._loop task = self._loop_task if loop is None or loop.is_closed(): return if task is not None and not task.done(): return created_task = loop.create_task(self._heartbeat_loop()) with self._lock: if self._loop is loop: self._loop_task = created_task else: created_task.cancel() async def _heartbeat_loop(self) -> None: control_plane.set_loop_running(True) try: while True: with self._lock: active_loop = self._loop if active_loop is None or active_loop is not asyncio.get_running_loop(): break try: await self._run_tick(force=False, reason="heartbeat") except Exception as error: logger.error("Autonomy heartbeat falhou: %s", error) control_plane.register_dispatch_failure(f"heartbeat_error:{error}") observability.record_error( category="autonomy_heartbeat", details=str(error), ) heartbeat_seconds = int( control_plane.get_config().get("heartbeat_interval_seconds", 60) ) await asyncio.sleep(max(15, heartbeat_seconds)) except asyncio.CancelledError: raise finally: control_plane.set_loop_running(False) async def _run_tick(self, *, force: bool, reason: str) -> dict[str, Any]: if self._tick_lock is None: self._tick_lock = asyncio.Lock() async with self._tick_lock: control_plane.touch_heartbeat() control_plane.register_tick(reason=reason) due_goals = control_plane.consume_due_goals(force=force) # Autonomia atualmente foca no canal principal configurado channel_id = control_plane.get_config().get("twitch_channel_login", "default") # Dynamic sentiment triggers (Phase 8) if not force: if sentiment_engine.should_trigger_anti_boredom(channel_id): due_goals.append( { "id": "dynamic-anti-boredom", "name": "Gatilho Anti-Tédio", "risk": RISK_AUTO_CHAT, "prompt": "O chat está pouco engajado ou 'morno'. Puxe um assunto interessante ou faça uma pergunta instigante para os viewers.", } ) if sentiment_engine.should_trigger_anti_confusion(channel_id): due_goals.append( { "id": "dynamic-anti-confusion", "name": "Gatilho Anti-Confusão", "risk": RISK_SUGGEST_STREAMER, "prompt": "O chat parece confuso com o que está acontecendo. Sugira ao streamer explicar melhor a situação atual ou o objetivo do jogo.", } ) # Auto-highlight on hype spikes. Self-gated (clip pipeline off -> # no-op) and throttled; never raises into the tick. try: from bot.highlight_runtime import highlight_runtime from bot.logic import BOT_BRAND, context_manager from bot.runtime_config import BYTE_VERSION, TWITCH_CHAT_MODE highlight_runtime.maybe_highlight( observability.snapshot( bot_brand=BOT_BRAND, bot_version=BYTE_VERSION, bot_mode=TWITCH_CHAT_MODE, stream_context=context_manager.get(channel_id), channel_id=channel_id, ) ) except Exception as error: logger.error("Highlight hook falhou: %s", error) processed: list[dict[str, Any]] = [] dispatcher = self._get_auto_chat_dispatcher() for goal in due_goals: result = await process_autonomy_goal(goal, dispatcher, channel_id=channel_id) safe_goal = goal if isinstance(goal, dict) else {} goal_id = str(safe_goal.get("id", "") or result.get("goal_id", "") or "").strip() goal_risk = ( str(safe_goal.get("risk", "") or result.get("risk", "") or "unknown") .strip() .lower() ) outcome = str(result.get("outcome", "") or "").strip().lower() evaluation = ( control_plane.register_goal_session_result( goal_id=goal_id, outcome=outcome, ) if goal_id else None ) if isinstance(evaluation, dict) and evaluation: observability.record_autonomy_goal( risk=goal_risk, outcome="kpi_met" if bool(evaluation.get("met")) else "kpi_missed", details=goal_id, channel_id=channel_id, ) result["session_kpi"] = evaluation processed.append(result) ops_playbooks = control_plane.run_ops_playbooks( channel_id=channel_id, reason=reason, ) return { "ok": True, "reason": reason, "force": bool(force), "due_goals": len(due_goals), "processed": processed, "ops_playbooks": ops_playbooks, "runtime": control_plane.runtime_snapshot(), } def _get_auto_chat_dispatcher(self) -> AutoChatDispatcher | None: with self._lock: return self._auto_chat_dispatcher def run_manual_tick(self, *, force: bool = True, reason: str = "manual") -> dict[str, Any]: coroutine = self._run_tick(force=force, reason=reason) with self._lock: loop = self._loop if loop is None or loop.is_closed() or not loop.is_running(): return asyncio.run(coroutine) future = asyncio.run_coroutine_threadsafe(coroutine, loop) try: return future.result(timeout=20.0) except FutureTimeoutError as error: future.cancel() raise TimeoutError("Autonomy tick timeout.") from error autonomy_runtime = AutonomyRuntime() __all__ = ["AutonomyRuntime", "autonomy_runtime"]