From f815fd5cd3a19db0ca117331f786674bb822c11a Mon Sep 17 00:00:00 2001 From: Homelab Cursor Date: Tue, 24 Mar 2026 14:11:38 +0100 Subject: [PATCH] =?UTF-8?q?Action=20Guard:=20Best=C3=A4tigung=20f=C3=BCr?= =?UTF-8?q?=20kritische=20Tools=20(savetv=5Frecord,=20savetv=5Fdownload,?= =?UTF-8?q?=20close=5Fissue,=20create=5Fissue)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - action_guard.py: wrap_handlers, is_confirmation, execute_pending - telegram_bot: Vor Ausführung Bestätigungsphrase (ja ausführen) erforderlich - abbruch beendet ausstehende Aktion --- homelab-ai-bot/action_guard.py | 132 +++++++++++++++++++++++++++++++++ homelab-ai-bot/telegram_bot.py | 27 +++++-- 2 files changed, 152 insertions(+), 7 deletions(-) create mode 100644 homelab-ai-bot/action_guard.py diff --git a/homelab-ai-bot/action_guard.py b/homelab-ai-bot/action_guard.py new file mode 100644 index 00000000..5cd4cc60 --- /dev/null +++ b/homelab-ai-bot/action_guard.py @@ -0,0 +1,132 @@ +"""Action Guard — Bestätigung für kritische Tool-Aufrufe. + +Kritische Tools werden nicht sofort ausgeführt. Stattdessen: +1. Bot antwortet: "Das würde X tun. Schreib 'ja ausführen' um zu bestätigen." +2. User schreibt Bestätigungsphrase +3. Dann wird die Aktion ausgeführt. +""" + +import logging +import re +from typing import Any, Callable, Optional + +log = logging.getLogger("action_guard") + +# Tools die eine explizite Bestätigung erfordern +CRITICAL_TOOLS = frozenset({ + "savetv_record", # Aufnahme anlegen + "savetv_download", # Film auf Festplatte + "close_issue", # Forgejo Issue schließen + "create_issue", # Forgejo Issue erstellen (kann Spam sein) +}) + +# Varianten der Bestätigungsphrase (normalisiert: lowercase, strip) +CONFIRM_PHRASES = [ + "ja ausführen", "ja ausfuehren", "ja bestätigen", "ja bestaetigen", + "bestätigen", "bestaetigen", "ja löschen", "ja loeschen", + "ja aufnehmen", "ja download", "ja runterladen", "bestätige", "bestaetige", +] + +# channel_key -> {tool_name, args, handler} +_pending: dict[str, dict] = {} + + +def _normalize(text: str) -> str: + t = (text or "").strip().lower() + for a, b in [("ä", "ae"), ("ö", "oe"), ("ü", "ue"), ("ß", "ss")]: + t = t.replace(a, b) + return re.sub(r"\s+", " ", t) + + +def is_confirmation(text: str) -> bool: + """Prüft ob der Text eine Bestätigungsphrase ist.""" + n = _normalize(text) + return any(phrase in n or n == phrase for phrase in CONFIRM_PHRASES) + + +def has_pending(channel_key: str) -> bool: + return channel_key in _pending + + +def clear_pending(channel_key: str) -> None: + _pending.pop(channel_key, None) + + +def get_pending_description(channel_key: str) -> Optional[str]: + p = _pending.get(channel_key) + if not p: + return None + return p.get("description", "kritische Aktion") + + +def execute_pending(channel_key: str) -> tuple[str, bool]: + """ + Führt die ausstehende Aktion aus. + Returns: (result_text, success) + """ + p = _pending.pop(channel_key, None) + if not p: + return "Keine ausstehende Aktion.", False + handler = p.get("handler") + args = p.get("args", {}) + tool_name = p.get("tool_name", "?") + if not handler: + return "Fehler: Handler fehlt.", False + try: + result = handler(**args) + log.info("Guard: %s ausgeführt nach Bestätigung", tool_name) + return str(result), True + except Exception as e: + log.exception("Guard: Fehler bei %s", tool_name) + return f"Fehler: {e}", False + + +def _describe_tool_call(tool_name: str, args: dict) -> str: + """Kurze Beschreibung der geplanten Aktion für den User.""" + if tool_name == "savetv_record": + tid = args.get("telecast_id", "?") + return f"Save.TV Aufnahme für TelecastId {tid} anlegen" + if tool_name == "savetv_download": + tid = args.get("telecast_id", "?") + title = args.get("title", "")[:40] + return f"Film (Id {tid}) auf Festplatte downloaden" + (f": {title}" if title else "") + if tool_name == "close_issue": + num = args.get("number", "?") + return f"Forgejo Issue #{num} schließen" + if tool_name == "create_issue": + title = (args.get("title") or "")[:50] + return f"Forgejo Issue erstellen: {title or '(ohne Titel)'}" + return f"Aktion {tool_name} ausführen" + + +def wrap_handlers(handlers: dict[str, Callable], channel_key: str) -> dict[str, Callable]: + """ + Gibt ein Wrapper-Dict zurück. Bei kritischen Tools: speichern statt ausführen. + """ + wrapped = {} + + for fn_name, handler in handlers.items(): + if fn_name not in CRITICAL_TOOLS: + wrapped[fn_name] = handler + continue + + def _capture(name: str, h: Callable): + def _wrapped(**kwargs): + if has_pending(channel_key): + return "Es gibt bereits eine ausstehende Aktion. Schreib 'ja ausführen' oder 'abbruch'." + desc = _describe_tool_call(name, kwargs) + _pending[channel_key] = { + "tool_name": name, + "args": kwargs, + "handler": h, + "description": desc, + } + return ( + f"⚠️ Das würde: {desc}.\n\n" + "Schreib 'ja ausführen' um zu bestätigen, oder 'abbruch' um abzubrechen." + ) + return _wrapped + + wrapped[fn_name] = _capture(fn_name, handler) + + return wrapped diff --git a/homelab-ai-bot/telegram_bot.py b/homelab-ai-bot/telegram_bot.py index ff7605ac..e0ed087f 100644 --- a/homelab-ai-bot/telegram_bot.py +++ b/homelab-ai-bot/telegram_bot.py @@ -105,6 +105,7 @@ import context import requests as _req import llm import memory_client +import action_guard import monitor import voice from core import config @@ -728,14 +729,24 @@ async def handle_message(update: Update, ctx: ContextTypes.DEFAULT_TYPE): if not text: return + channel_key = str(update.effective_chat.id) if text.strip().lower() in ("abbruch", "stop", "stopp", "cancel"): - chat_id = update.effective_chat.id - task = ACTIVE_LLM_TASKS.get(chat_id) - if task and not task.done(): - task.cancel() - await update.message.reply_text("🛑 Abgebrochen.") + if action_guard.has_pending(channel_key): + action_guard.clear_pending(channel_key) + await update.message.reply_text("🛑 Ausstehende Aktion abgebrochen.") else: - await update.message.reply_text("Kein laufender Suchlauf.") + chat_id = update.effective_chat.id + task = ACTIVE_LLM_TASKS.get(chat_id) + if task and not task.done(): + task.cancel() + await update.message.reply_text("🛑 Abgebrochen.") + else: + await update.message.reply_text("Kein laufender Suchlauf.") + return + + if action_guard.is_confirmation(text) and action_guard.has_pending(channel_key): + result, ok = action_guard.execute_pending(channel_key) + await update.message.reply_text(("✅ " if ok else "") + result[:4000]) return cmd = BUTTON_MAP.get(text) @@ -761,7 +772,9 @@ async def handle_message(update: Update, ctx: ContextTypes.DEFAULT_TYPE): try: context.last_suggest_result = {"type": None} context.set_source_type("telegram_text") - handlers = context.get_tool_handlers(session_id=session_id) + handlers = action_guard.wrap_handlers( + context.get_tool_handlers(session_id=session_id), channel_key + ) llm_task = asyncio.create_task( asyncio.to_thread(llm.ask_with_tools, text, handlers, session_id=session_id) )