Action Guard: Bestätigung für kritische Tools (savetv_record, savetv_download, close_issue, create_issue)

- 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

@@ -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

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
+        if action_guard.has_pending(channel_key):
-        task = ACTIVE_LLM_TASKS.get(chat_id)
+            action_guard.clear_pending(channel_key)
-        if task and not task.done():
+            await update.message.reply_text("🛑 Ausstehende Aktion abgebrochen.")
-            task.cancel()
-            await update.message.reply_text("🛑 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)
         )