"""Sliding-window budget for bot-authored inbound messages. ``{PLATFORM}_ALLOW_BOTS`` only decides admission, so two Hermes profiles replying to each other never stop. The guard counts admitted bot messages per conversation and drops further ones for ``cooldown_seconds`` once ``max_events`` land inside ``window_seconds``. Settings: config.yaml ``gateway.bot_loop_guard``. """ from __future__ import annotations import threading import time from collections import deque from dataclasses import dataclass from typing import Callable, Deque, Dict, Hashable, Tuple __all__ = ["BotLoopGuard", "BotLoopGuardSettings", "load_settings", "settings_from_config"] _TRUTHY = frozenset({"true", "1", "yes", "on"}) _FALSY = frozenset({"false", "0", "no", "off"}) @dataclass(frozen=True) class BotLoopGuardSettings: enabled: bool = True max_events: int = 20 window_seconds: float = 300.0 cooldown_seconds: float = 600.0 def _as_bool(raw, default: bool) -> bool: if isinstance(raw, bool): return raw text = str(raw).strip().lower() if raw is not None else "" if text in _TRUTHY: return True if text in _FALSY: return False return default def _as_positive(raw, default: float) -> float: if isinstance(raw, bool): return default try: value = float(raw) except (TypeError, ValueError): return default return value if value > 0 else default def _as_positive_int(raw, default: int) -> int: value = _as_positive(raw, 0.0) return int(value) if value >= 1 and value == int(value) else default def settings_from_config(cfg) -> BotLoopGuardSettings: """Read ``gateway.bot_loop_guard`` from a loaded config dict. Unusable values keep the default.""" from hermes_cli.config import cfg_get block = cfg_get(cfg, "gateway", "bot_loop_guard", default=None) if not isinstance(block, dict): return BotLoopGuardSettings() defaults = BotLoopGuardSettings() return BotLoopGuardSettings( enabled=_as_bool(block.get("enabled"), defaults.enabled), max_events=_as_positive_int(block.get("max_events"), defaults.max_events), window_seconds=_as_positive(block.get("window_seconds"), defaults.window_seconds), cooldown_seconds=_as_positive(block.get("cooldown_seconds"), defaults.cooldown_seconds), ) def load_settings() -> BotLoopGuardSettings: """Settings from the live config.yaml. Defaults when the config cannot be read.""" try: from hermes_cli.config import load_config_readonly return settings_from_config(load_config_readonly()) except Exception: return BotLoopGuardSettings() class BotLoopGuard: """Per-conversation sliding window with a cooldown once the budget trips. Thread-safe. Settings are re-read on every call so a config.yaml edit takes effect without a restart.""" def __init__( self, settings: Callable[[], BotLoopGuardSettings] = load_settings, clock: Callable[[], float] = time.monotonic, ) -> None: self._settings = settings self._clock = clock self._lock = threading.Lock() self._events: Dict[Hashable, Deque[float]] = {} self._cooldown_until: Dict[Hashable, float] = {} self._last_sweep = 0.0 @property def tracked_conversations(self) -> int: with self._lock: return len(self._events) def blocked(self, conversation: Hashable) -> bool: """True while ``conversation`` is cooling down. Reads only, so callers may ask as often as they like.""" if not self._settings().enabled: return False with self._lock: return self._cooldown_until.get(conversation, 0.0) > self._clock() def admit(self, conversation: Hashable) -> Tuple[bool, str]: """Count one admitted bot-authored message for ``conversation``. Returns ``(allowed, state)``; state is ``disabled``, ``ok``, ``tripped`` (this message started the cooldown) or ``cooldown``.""" settings = self._settings() if not settings.enabled: return True, "disabled" now = self._clock() with self._lock: self._sweep(now, settings) if self._cooldown_until.get(conversation, 0.0) > now: return False, "cooldown" events = self._events.setdefault(conversation, deque()) cutoff = now - settings.window_seconds while events and events[0] <= cutoff: events.popleft() if len(events) >= settings.max_events: self._cooldown_until[conversation] = now + settings.cooldown_seconds events.clear() return False, "tripped" events.append(now) return True, "ok" def _sweep(self, now: float, settings: BotLoopGuardSettings) -> None: """Drop idle conversations at most once per window so memory stays bounded.""" if now - self._last_sweep < settings.window_seconds: return self._last_sweep = now idle_cutoff = now - settings.window_seconds for key in [k for k, dq in self._events.items() if not dq or dq[-1] <= idle_cutoff]: del self._events[key] for key in [k for k, until in self._cooldown_until.items() if until <= now]: del self._cooldown_until[key]