Spaces:
Sleeping
Sleeping
| from typing import Any | |
| from bot.control_plane_actions import ControlPlaneActionQueue | |
| from bot.control_plane_config import ControlPlaneConfigRuntime | |
| from bot.control_plane_constants import ( | |
| RISK_AUTO_CHAT, | |
| RISK_CLIP_CANDIDATE, | |
| RISK_MODERATION_ACTION, | |
| RISK_SUGGEST_STREAMER, | |
| SUPPORTED_DECISIONS, | |
| SUPPORTED_RISK_LEVELS, | |
| ) | |
| from bot.ops_playbooks import OpsPlaybookRuntime | |
| class ControlPlaneState: | |
| MAX_ACTION_ITEMS = 600 | |
| def __init__(self) -> None: | |
| self._config_runtime = ControlPlaneConfigRuntime() | |
| self._action_queue = ControlPlaneActionQueue(max_items=self.MAX_ACTION_ITEMS) | |
| self._ops_playbooks = OpsPlaybookRuntime() | |
| def _action_ignore_after_seconds(self) -> int: | |
| return self._config_runtime.action_ignore_after_seconds() | |
| def _get_action_for_ops_playbook( | |
| self, | |
| action_id: str, | |
| *, | |
| timestamp: float | None = None, | |
| ) -> dict[str, Any] | None: | |
| return self._action_queue.get_action( | |
| action_id, | |
| ignore_after_seconds=self._action_ignore_after_seconds(), | |
| timestamp=timestamp, | |
| ) | |
| def _ops_playbook_metrics(self, *, timestamp: float | None = None) -> dict[str, Any]: | |
| queue_runtime = self._action_queue.runtime_snapshot( | |
| ignore_after_seconds=self._action_ignore_after_seconds(), | |
| timestamp=timestamp, | |
| ) | |
| queue = queue_runtime.get("queue", {}) | |
| queue_window_60m = queue_runtime.get("queue_window_60m", {}) | |
| return { | |
| "queue_pending": int(queue.get("pending", 0)), | |
| "queue_ignored_rate_60m": float(queue_window_60m.get("ignored_rate", 0.0)), | |
| "queue_decisions_total_60m": int(queue_window_60m.get("decisions_total", 0)), | |
| "queue_target_pending": 3, | |
| } | |
| def get_config(self) -> dict[str, Any]: | |
| return self._config_runtime.get_config() | |
| def update_config(self, payload: dict[str, Any]) -> dict[str, Any]: | |
| return self._config_runtime.update_config(payload) | |
| def suspend_agent( | |
| self, | |
| *, | |
| reason: str = "manual_dashboard", | |
| timestamp: float | None = None, | |
| ) -> dict[str, Any]: | |
| return self._config_runtime.suspend_agent(reason=reason, timestamp=timestamp) | |
| def resume_agent( | |
| self, | |
| *, | |
| reason: str = "manual_dashboard", | |
| timestamp: float | None = None, | |
| ) -> dict[str, Any]: | |
| return self._config_runtime.resume_agent(reason=reason, timestamp=timestamp) | |
| def set_loop_running(self, running: bool) -> None: | |
| self._config_runtime.set_loop_running(running) | |
| def touch_heartbeat(self, timestamp: float | None = None) -> None: | |
| self._config_runtime.touch_heartbeat(timestamp=timestamp) | |
| def register_tick(self, reason: str, timestamp: float | None = None) -> None: | |
| self._config_runtime.register_tick(reason=reason, timestamp=timestamp) | |
| def register_goal_run( | |
| self, | |
| goal_id: str, | |
| risk: str, | |
| timestamp: float | None = None, | |
| ) -> None: | |
| self._config_runtime.register_goal_run( | |
| goal_id=goal_id, | |
| risk=risk, | |
| timestamp=timestamp, | |
| ) | |
| def register_budget_block(self, reason: str, timestamp: float | None = None) -> None: | |
| self._config_runtime.register_budget_block(reason=reason, timestamp=timestamp) | |
| def register_dispatch_failure(self, reason: str, timestamp: float | None = None) -> None: | |
| self._config_runtime.register_dispatch_failure(reason=reason, timestamp=timestamp) | |
| def register_goal_session_result( | |
| self, | |
| *, | |
| goal_id: str, | |
| outcome: str, | |
| observed_value: float | None = None, | |
| details: str = "", | |
| timestamp: float | None = None, | |
| ) -> dict[str, Any] | None: | |
| return self._config_runtime.register_goal_session_result( | |
| goal_id=goal_id, | |
| outcome=outcome, | |
| observed_value=observed_value, | |
| details=details, | |
| timestamp=timestamp, | |
| ) | |
| def can_send_auto_chat( | |
| self, timestamp: float | None = None | |
| ) -> tuple[bool, str, dict[str, int]]: | |
| return self._config_runtime.can_send_auto_chat(timestamp=timestamp) | |
| def register_auto_chat_sent(self, timestamp: float | None = None) -> None: | |
| self._config_runtime.register_auto_chat_sent(timestamp=timestamp) | |
| def consume_due_goals( | |
| self, *, force: bool = False, timestamp: float | None = None | |
| ) -> list[dict[str, Any]]: | |
| return self._config_runtime.consume_due_goals(force=force, timestamp=timestamp) | |
| def enqueue_action( | |
| self, | |
| *, | |
| kind: str, | |
| risk: str, | |
| title: str, | |
| body: str, | |
| payload: dict[str, Any] | None = None, | |
| created_by: str = "autonomy", | |
| timestamp: float | None = None, | |
| ) -> dict[str, Any]: | |
| return self._action_queue.enqueue_action( | |
| kind=kind, | |
| risk=risk, | |
| title=title, | |
| body=body, | |
| payload=payload, | |
| created_by=created_by, | |
| timestamp=timestamp, | |
| ) | |
| def decide_action( | |
| self, | |
| *, | |
| action_id: str, | |
| decision: str, | |
| note: str = "", | |
| decided_by: str = "dashboard", | |
| timestamp: float | None = None, | |
| ) -> dict[str, Any]: | |
| return self._action_queue.decide_action( | |
| action_id=action_id, | |
| decision=decision, | |
| note=note, | |
| decided_by=decided_by, | |
| timestamp=timestamp, | |
| ) | |
| def list_actions( | |
| self, | |
| *, | |
| status: str | None = None, | |
| limit: int = 80, | |
| timestamp: float | None = None, | |
| ) -> dict[str, Any]: | |
| return self._action_queue.list_actions( | |
| status=status, | |
| limit=limit, | |
| ignore_after_seconds=self._action_ignore_after_seconds(), | |
| timestamp=timestamp, | |
| ) | |
| def ops_playbooks_snapshot( | |
| self, | |
| *, | |
| channel_id: str = "default", | |
| timestamp: float | None = None, | |
| ) -> dict[str, Any]: | |
| return self._ops_playbooks.reconcile( | |
| channel_id=channel_id, | |
| metrics=self._ops_playbook_metrics(timestamp=timestamp), | |
| get_action=self._get_action_for_ops_playbook, | |
| enqueue_action=self._action_queue.enqueue_action, | |
| timestamp=timestamp, | |
| ) | |
| def run_ops_playbooks( | |
| self, | |
| *, | |
| channel_id: str = "default", | |
| reason: str = "autonomy_tick", | |
| timestamp: float | None = None, | |
| ) -> dict[str, Any]: | |
| return self._ops_playbooks.evaluate( | |
| channel_id=channel_id, | |
| metrics=self._ops_playbook_metrics(timestamp=timestamp), | |
| trigger_reason=reason, | |
| get_action=self._get_action_for_ops_playbook, | |
| enqueue_action=self._action_queue.enqueue_action, | |
| timestamp=timestamp, | |
| ) | |
| def trigger_ops_playbook( | |
| self, | |
| *, | |
| playbook_id: str, | |
| channel_id: str = "default", | |
| reason: str = "manual_dashboard", | |
| force: bool = False, | |
| timestamp: float | None = None, | |
| ) -> dict[str, Any]: | |
| return self._ops_playbooks.trigger( | |
| playbook_id=playbook_id, | |
| channel_id=channel_id, | |
| reason=reason, | |
| metrics=self._ops_playbook_metrics(timestamp=timestamp), | |
| get_action=self._get_action_for_ops_playbook, | |
| enqueue_action=self._action_queue.enqueue_action, | |
| force=force, | |
| timestamp=timestamp, | |
| ) | |
| def build_capabilities(self, *, bot_mode: str) -> dict[str, Any]: | |
| safe_mode = (bot_mode or "eventsub").strip().lower() or "eventsub" | |
| channel_enabled = safe_mode == "irc" | |
| if channel_enabled: | |
| channel_reason = "Controle de canais ativo no runtime IRC." | |
| supported_actions = ["list", "join", "part"] | |
| else: | |
| channel_reason = ( | |
| "Channel control de runtime so funciona em TWITCH_CHAT_MODE=irc. " | |
| "No modo eventsub, join/part ficam bloqueados." | |
| ) | |
| supported_actions = [] | |
| config = self.get_config() | |
| return { | |
| "channel_control": { | |
| "enabled": channel_enabled, | |
| "reason": channel_reason, | |
| "supported_actions": supported_actions, | |
| }, | |
| "autonomy": { | |
| "enabled": True, | |
| "reason": "Autonomia com agenda e budget anti-spam disponivel.", | |
| "active": bool(config.get("autonomy_enabled", False)), | |
| "suspended": bool(config.get("agent_suspended", False)), | |
| "suspendable": True, | |
| }, | |
| "risk_actions": { | |
| "enabled": True, | |
| "reason": "Fluxo por risco com fila e decisao approve/reject.", | |
| "levels": sorted(SUPPORTED_RISK_LEVELS), | |
| }, | |
| "ops_playbooks": { | |
| "enabled": True, | |
| "reason": "Playbooks deterministicos com state machine e trilha auditavel.", | |
| "surface": "risk_queue_panel", | |
| }, | |
| "response_contract": { | |
| "max_messages": 1, | |
| "max_lines": 4, | |
| }, | |
| "clip_pipeline": { | |
| "enabled": config.get("clip_pipeline_enabled", False), | |
| "modes": ["live", "vod"], | |
| "default_mode": config.get("clip_mode_default", "live"), | |
| }, | |
| } | |
| def runtime_snapshot(self, *, timestamp: float | None = None) -> dict[str, Any]: | |
| runtime = self._config_runtime.runtime_base_snapshot(timestamp=timestamp) | |
| queue_runtime = self._action_queue.runtime_snapshot( | |
| ignore_after_seconds=self._action_ignore_after_seconds(), | |
| timestamp=timestamp, | |
| ) | |
| runtime["queue"] = queue_runtime.get("queue", {}) | |
| runtime["queue_window_60m"] = queue_runtime.get("queue_window_60m", {}) | |
| runtime["ops_playbooks"] = self.ops_playbooks_snapshot(timestamp=timestamp) | |
| return runtime | |
| control_plane = ControlPlaneState() | |
| __all__ = [ | |
| "RISK_AUTO_CHAT", | |
| "RISK_CLIP_CANDIDATE", | |
| "RISK_MODERATION_ACTION", | |
| "RISK_SUGGEST_STREAMER", | |
| "SUPPORTED_DECISIONS", | |
| "SUPPORTED_RISK_LEVELS", | |
| "ControlPlaneState", | |
| "control_plane", | |
| ] | |