Browser_Agent/agent/tools.py

# agent/tools.py
import logging
import asyncio
import hashlib
import json
import uuid
from datetime import datetime, timedelta, timezone
from typing import Dict, Any
from urllib.parse import urlparse
from playwright.async_api import Page

from agent.handoff import build_handoff_result, normalize_handoff_url
from agent.tool_registry import TOOL_SPECS
from agent.vision_subagent import VisionCheckError, vision_subagent

logger = logging.getLogger(__name__)


def _page_revision(page: Page) -> int:
    return int(getattr(page, "__dict__", {}).get("_gravity_snapshot_revision", 0))


def _advance_revision(page: Page) -> int:
    revision = _page_revision(page) + 1
    setattr(page, "_gravity_snapshot_revision", revision)
    return revision


def _stale_snapshot(page: Page, func_args: Dict[str, Any]) -> bool:
    supplied = func_args.get("snapshot_revision")
    return supplied is not None and int(supplied) != _page_revision(page)


async def _human_mouse_move(page: Page, x: float, y: float) -> None:
    """Move through Playwright with a short, bounded human-like path."""
    await page.mouse.move(x, y, steps=6)


async def _human_type(page: Page, value: str, selector: str) -> None:
    """Focus and fill an annotated field without Playwright's 30s default wait."""
    field = page.locator(selector)
    await field.click(timeout=3000)
    await field.fill(value, timeout=3000)

async def _execute_tool_impl(func_name: str, func_args: Dict[str, Any], page: Page, task=None) -> Dict[str, Any]:
    """
    Führt ein Tool basierend auf den von der AI gelieferten Argumenten aus.
    """
    try:
        if func_name == "navigate":
            url = func_args.get("url")
            if not url:
                return {"error": "Missing 'url' argument."}
            # --- NEU: SSRF Validation ---
            try:
                from security.ssrf_guard import validate_url_async
                url = await validate_url_async(url)
            except Exception as e:
                # SecurityError (oder Import Fehler) loggen und Agent blockieren
                logger.warning(f"Sicherheitsverstoß beim Navigieren zu {url}: {e}")
                return {"error": f"Sicherheitsverstoß blockiert: {e}"}

            # Attached sessions are origin-scoped. Never spray every cookie
            # from the chat session at an unrelated destination.
            if task is not None and isinstance(getattr(task, "params", None), dict):
                host = (urlparse(url).hostname or "").lower()
                injected = set(task.params.get("_injected_cookie_origins") or [])
                if host not in injected:
                    scoped_cookies = []
                    for site in task.params.get("exported_sites") or []:
                        if not isinstance(site, dict):
                            continue
                        domain = str(site.get("domain") or "").lstrip(".").lower()
                        if not domain or (host != domain and not host.endswith(f".{domain}")):
                            continue
                        for cookie in site.get("cookies") or []:
                            if not isinstance(cookie, dict) or not cookie.get("name"):
                                continue
                            normalized = {
                                "name": str(cookie.get("name")),
                                "value": str(cookie.get("value", "")),
                                "domain": str(cookie.get("domain") or domain),
                                "path": str(cookie.get("path") or "/"),
                            }
                            if "isSecure" in cookie or "secure" in cookie:
                                normalized["secure"] = bool(cookie.get("isSecure", cookie.get("secure")))
                            if "isHttpOnly" in cookie or "httpOnly" in cookie:
                                normalized["httpOnly"] = bool(cookie.get("isHttpOnly", cookie.get("httpOnly")))
                            scoped_cookies.append(normalized)
                    if scoped_cookies:
                        await page.context.add_cookies(scoped_cookies)
                    injected.add(host)
                    task.params["_injected_cookie_origins"] = sorted(injected)

            logger.info(f"Tool: Navigating to {url}")
            await page.goto(url, wait_until="domcontentloaded", timeout=30000)
            revision = _advance_revision(page)
            return {"status": "success", "message": f"Navigated to {url}", "snapshot_revision": revision}

        elif func_name == "go_back":
            await page.go_back(wait_until="domcontentloaded", timeout=30000)
            return {"status": "success", "snapshot_revision": _advance_revision(page)}

        elif func_name == "go_forward":
            await page.go_forward(wait_until="domcontentloaded", timeout=30000)
            return {"status": "success", "snapshot_revision": _advance_revision(page)}

        elif func_name == "reload":
            await page.reload(wait_until="domcontentloaded", timeout=30000)
            return {"status": "success", "snapshot_revision": _advance_revision(page)}

        elif func_name == "wait_for":
            selector = str(func_args.get("selector") or "")
            timeout_ms = min(max(int(func_args.get("timeout_ms") or 5000), 100), 15000)
            if selector:
                await page.locator(selector).first.wait_for(state="visible", timeout=timeout_ms)
            else:
                await page.wait_for_load_state("networkidle", timeout=timeout_ms)
            return {"status": "success", "snapshot_revision": _page_revision(page)}
            
        elif func_name == "request_user_permission":
            reason = func_args.get("reason")
            if not reason:
                return {"error": "Missing 'reason' argument."}
            if task is None or not isinstance(getattr(task, "params", None), dict):
                return {"error": "No task context available for confirmation."}
            logger.info(f"Tool: Requesting permission: {reason}")
            confirmation_id = f"confirm_{uuid.uuid4().hex}"
            page_url = str(page.url)
            parsed = urlparse(page_url)
            origin = f"{parsed.scheme}://{parsed.netloc}"
            expires_at = datetime.now(timezone.utc) + timedelta(minutes=5)
            digest = hashlib.sha256(
                f"{origin}|submit|{reason}".encode("utf-8")
            ).hexdigest()
            task.params["_confirmation_checkpoint"] = {
                "id": confirmation_id,
                "url": page_url,
                "cookies": await page.context.cookies(),
                "origin": origin,
                "action_digest": digest,
                "expires_at": expires_at.isoformat(),
            }
            return {
                "status": "confirmation_required",
                "message": reason,
                "confirmation": {
                    "id": confirmation_id,
                    "reason": reason,
                    "origin": origin,
                    "risk": "external_effect",
                    "action": "submit",
                    "action_digest": digest,
                    "expires_at": expires_at.isoformat(),
                },
            }
            
        elif func_name == "request_temp_mail_bypass":
            email = func_args.get("email_address", "")
            if not email:
                return {"error": "Missing 'email_address' argument."}
            logger.info("Deprecated temp-mail mutation bypass was rejected")
            return {
                "status": "error",
                "error_code": "confirmation_required",
                "error": (
                    "Auch Registrierungen mit temporärer E-Mail benötigen vor dem "
                    "Absenden eine konkrete Nutzerbestätigung."
                ),
            }
                
        elif func_name == "request_browser_handoff":
            reason = func_args.get("reason", "Handoff requested")
            cookies = await page.context.cookies()
            url = normalize_handoff_url(str(page.url))
            account_context = None
            if task is not None and isinstance(getattr(task, "params", None), dict):
                candidate = task.params.get("_account_creation_context")
                if isinstance(candidate, dict):
                    account_context = candidate
                    url = normalize_handoff_url(
                        str(candidate.get("target_url") or page.url)
                    )
            form_data = None
            if account_context:
                form_data = {
                    "email": str(account_context.get("email", "")),
                    "username": str(account_context.get("username", "")),
                    "password": str(account_context.get("password", "")),
                }
            logger.info(f"Tool: Browser Handoff requested for {url}")
            result = build_handoff_result(
                url,
                cookies,
                reason,
                form_data=form_data,
                handoff_type=(
                    "registration_captcha" if account_context else "browser_challenge"
                ),
                completion_policy=(
                    "registration_submitted" if account_context else "challenge_completed"
                ),
            )
            logger.info(
                "Browser-Handoff vorbereitet: %d von %d Cookies übernommen",
                result["cookie_count"],
                len(cookies),
            )
            return result
            
        elif func_name == "inject_cookies":
            cookies_json = func_args.get("cookies_json", "[]")
            resume_url = func_args.get("resume_url", "")
            import json
            try:
                cookies = json.loads(cookies_json)
                await page.context.add_cookies(cookies)
                if resume_url:
                    await page.goto(resume_url, wait_until="domcontentloaded")
                return {"status": "success", "message": "Cookies erfolgreich injiziert. Session wiederhergestellt."}
            except Exception as e:
                return {"error": f"Fehler beim Injizieren der Cookies: {e}"}

        elif func_name == "search_web":
            query = func_args.get("query")
            if not query:
                return {"error": "Missing 'query' argument."}
            logger.info(f"Tool: Searching web for: {query}")
            try:
                from duckduckgo_search import DDGS
                with DDGS() as ddgs:
                    results = list(ddgs.text(query, max_results=3))
                return {"status": "success", "results": results}
            except Exception as e:
                logger.error(f"Search error: {e}")
                return {"error": f"Search failed: {e}"}

        elif func_name == "create_temp_account":
            target_url = func_args.get("target_url")
            if not target_url:
                return {"error": "Missing 'target_url' argument."}
            provided_email = func_args.get("provided_email", "").strip() or None
            logger.info(f"Tool: Creating temp account for {target_url} (provided_email={provided_email})")
            try:
                from workflows.account_creator import AccountCreatorWorkflow
                workflow = AccountCreatorWorkflow(page)
                result = await workflow.run(target_url, task=task, provided_email=provided_email)
                if not result.get("success", False) and result.get("status") != "handoff_required":
                    return {
                        "status": "error",
                        "success": False,
                        "error": result.get("error", "Registrierung fehlgeschlagen."),
                        "suggestion": result.get("suggestion", "Bitte nutze Google/GitHub OAuth oder eine echte E-Mail."),
                        "email": result.get("email", "")
                    }
                return result
            except Exception as e:
                logger.error(f"Error in create_temp_account workflow: {e}")
                return {"status": "error", "success": False, "error": f"Workflow failed: {e}"}

        elif func_name == "click":
            node_id = func_args.get("node_id")
            if not node_id:
                return {"error": "Missing 'node_id' argument."}
            if _stale_snapshot(page, func_args):
                return {
                    "status": "error",
                    "error_code": "stale_snapshot",
                    "error": "Die Seite hat sich geändert. Beobachte sie neu und versuche den Klick einmal erneut.",
                }
                
            logger.info(f"Tool: Clicking node {node_id}")
            
            selector = f"[data-ax-node-id='{node_id}']"
            try:
                is_form_submit = await page.evaluate(f'''() => {{
                        const el = document.querySelector("{selector}");
                        if (!el) return false;
                        
                        // 1. Ist es ein Submit Button?
                        if (el.tagName.toLowerCase() === 'button' && el.type === 'submit') return true;
                        if (el.tagName.toLowerCase() === 'input' && el.type === 'submit') return true;
                        
                        // 2. Ist es in einer Form und sendet diese durch Klick ab?
                        const form = el.closest('form');
                        if (form && el.tagName.toLowerCase() === 'button') return true;
                        
                        // 3. Text Heuristik (SPA Fallback)
                        const text = (el.textContent || el.value || el.ariaLabel || "").toLowerCase();
                        const submitWords = ['submit', 'login', 'anmelden', 'register', 'registrieren', 'kaufen', 'senden'];
                        if (submitWords.some(word => text.includes(word))) return true;
                        
                        return false;
                    }}''')
            except Exception as e:
                logger.warning(f"Error during submit check: {e}")
                is_form_submit = False

            approved = None
            confirmation_id = str(func_args.get("confirmation_id") or "")
            if task is not None and isinstance(getattr(task, "params", None), dict):
                candidate = task.params.get("_approved_confirmation")
                if isinstance(candidate, dict):
                    approved = candidate
            if is_form_submit:
                parsed = urlparse(str(page.url))
                current_origin = f"{parsed.scheme}://{parsed.netloc}"
                expires_at = None
                if approved and approved.get("expires_at"):
                    try:
                        expires_at = datetime.fromisoformat(str(approved["expires_at"]))
                    except ValueError:
                        expires_at = None
                valid_confirmation = bool(
                    approved
                    and confirmation_id
                    and confirmation_id == approved.get("id")
                    and approved.get("origin") == current_origin
                    and approved.get("consumed") is not True
                    and expires_at
                    and expires_at > datetime.now(timezone.utc)
                )
                if not valid_confirmation:
                    return {
                        "status": "error",
                        "error_code": "confirmation_required",
                        "error": (
                            "Diese Aktion würde ein Formular absenden. Fordere zuerst "
                            "eine konkrete Nutzerbestätigung an."
                        ),
                    }

            # Human-like mouse interaction
            try:
                box = await page.evaluate(f'''() => {{
                    const el = document.querySelector("{selector}");
                    if (!el) return null;
                    const r = el.getBoundingClientRect();
                    return {{ x: r.x + r.width/2, y: r.y + r.height/2 }};
                }}''')
                if box and box.get("x") is not None and box.get("y") is not None:
                    await _human_mouse_move(page, box["x"], box["y"])
                    await page.mouse.click(box["x"], box["y"])
                else:
                    return {
                        "error": (
                            f"Element '{node_id}' ist nicht mehr vorhanden oder sichtbar. "
                            "Lies den DOM erneut und nutze die aktuelle id."
                        )
                    }
            except Exception as exc:
                return {"error": f"Klick auf Element '{node_id}' fehlgeschlagen: {exc}"}
            
            if is_form_submit and approved:
                approved["consumed"] = True
                task.params.pop("_approved_confirmation", None)
                
            return {
                "status": "success",
                "message": f"Clicked element {node_id}",
                "snapshot_revision": _advance_revision(page),
            }
            
        elif func_name == "define_plan":
            steps = func_args.get("steps", [])
            if not task:
                return {"error": "No task context available to define plan."}
            now_str = datetime.now().strftime("%H:%M:%S")
            task.steps = [
                {
                    "title": s,
                    "subtitle": "Warte auf Ausführung..." if i > 0 else "Wird ausgeführt...",
                    "status": "active" if i == 0 else "pending",
                    "time": now_str
                }
                for i, s in enumerate(steps)
            ]
            logger.info(f"Tool: Plan definiert mit {len(steps)} Schritten (Schritt 1 aktiv)")
            return {"status": "success", "message": f"Plan mit {len(steps)} Schritten definiert. Schritt 1 ist jetzt aktiv."}

        elif func_name == "update_step_status":
            index = func_args.get("step_index")
            status = func_args.get("status")
            if not task or not task.steps:
                return {"error": "No plan defined."}
            if index is None or index < 0 or index >= len(task.steps):
                return {"error": "Invalid step_index."}
            task.steps[index]["status"] = status
            # Falls ein Schritt auf done gesetzt wird und der nächste noch pending ist, setze den nächsten auf active
            if status == "done" and index + 1 < len(task.steps) and task.steps[index + 1].get("status") == "pending":
                task.steps[index + 1]["status"] = "active"
                task.steps[index + 1]["subtitle"] = "Wird ausgeführt..."
            logger.info(f"Tool: Schritt {index} auf {status} gesetzt")
            return {"status": "success", "message": f"Schritt {index} aktualisiert."}

        elif func_name == "task_complete":
            result = func_args.get("result", "")
            logger.info(f"Tool: Task completed with result: {result[:50]}")
            return {"status": "success", "message": "Task complete."}
            
        elif func_name == "type_text":
            node_id = func_args.get("node_id")
            value = func_args.get("value")
            if not node_id or not value:
                return {"error": "Missing 'node_id' or 'value' argument."}
            if _stale_snapshot(page, func_args):
                return {
                    "status": "error",
                    "error_code": "stale_snapshot",
                    "error": "Das Eingabefeld stammt aus einem alten Seitensnapshot. Beobachte die Seite neu.",
                }
            logger.info(f"Tool: Typing into node {node_id}")
            selector = f"[data-ax-node-id='{node_id}']"
            # Human-like typing with natural jitter
            try:
                await _human_type(page, value, selector=selector)
            except Exception as e:
                logger.warning(f"Human typing fallback: {e}")
                await page.fill(selector, value, timeout=3000)
            return {
                "status": "success",
                "message": f"Typed text into {node_id}",
                "snapshot_revision": _advance_revision(page),
            }

        elif func_name == "select_option":
            node_id = str(func_args.get("node_id") or "")
            value = str(func_args.get("value") or "")
            if not node_id or not value:
                return {"error": "Missing 'node_id' or 'value' argument."}
            if _stale_snapshot(page, func_args):
                return {"status": "error", "error_code": "stale_snapshot", "error": "Observe the page again."}
            selected = await page.locator(f"[data-ax-node-id='{node_id}']").select_option(value=value, timeout=3000)
            return {"status": "success", "selected": selected, "snapshot_revision": _advance_revision(page)}

        elif func_name == "set_checked":
            node_id = str(func_args.get("node_id") or "")
            checked = func_args.get("checked") is True
            if not node_id:
                return {"error": "Missing 'node_id' argument."}
            if _stale_snapshot(page, func_args):
                return {"status": "error", "error_code": "stale_snapshot", "error": "Observe the page again."}
            locator = page.locator(f"[data-ax-node-id='{node_id}']")
            await locator.set_checked(checked, timeout=3000)
            return {"status": "success", "checked": checked, "snapshot_revision": _advance_revision(page)}

        elif func_name == "press_key":
            key = str(func_args.get("key") or "")
            node_id = str(func_args.get("node_id") or "")
            if not key:
                return {"error": "Missing 'key' argument."}
            if node_id:
                await page.locator(f"[data-ax-node-id='{node_id}']").press(key, timeout=3000)
            else:
                await page.keyboard.press(key)
            return {"status": "success", "snapshot_revision": _advance_revision(page)}

        elif func_name == "hover":
            node_id = str(func_args.get("node_id") or "")
            if not node_id:
                return {"error": "Missing 'node_id' argument."}
            await page.locator(f"[data-ax-node-id='{node_id}']").hover(timeout=3000)
            return {"status": "success", "snapshot_revision": _page_revision(page)}

        elif func_name == "scroll":
            delta_y = min(max(int(func_args.get("delta_y") or 600), -4000), 4000)
            await page.mouse.wheel(0, delta_y)
            return {"status": "success", "snapshot_revision": _page_revision(page)}

        elif func_name == "list_tabs":
            pages = page.context.pages
            return {
                "status": "success",
                "tabs": [
                    {"index": index, "url": tab.url, "title": await tab.title()}
                    for index, tab in enumerate(pages)
                ],
            }

        elif func_name == "switch_tab":
            index = int(func_args.get("index", -1))
            pages = page.context.pages
            if index < 0 or index >= len(pages):
                return {"status": "error", "error_code": "tab_not_found", "error": "Tab index is invalid."}
            target = pages[index]
            await target.bring_to_front()
            root_page = getattr(page, "__dict__", {}).get("_gravity_root_page", page)
            setattr(target, "_gravity_root_page", root_page)
            setattr(root_page, "_gravity_active_page", target)
            return {"status": "success", "url": target.url, "snapshot_revision": _advance_revision(target)}

        elif func_name == "close_tab":
            index = int(func_args.get("index", -1))
            pages = page.context.pages
            if index < 0 or index >= len(pages) or pages[index] is page:
                return {"status": "error", "error_code": "tab_not_closable", "error": "The requested tab cannot be closed."}
            await pages[index].close()
            return {"status": "success", "remaining_tabs": len(page.context.pages)}

        elif func_name == "extract_structured":
            fields = func_args.get("fields") or {}
            if not isinstance(fields, dict) or not fields:
                return {"error": "Missing 'fields' selector map."}
            extracted = {}
            for name, selector in list(fields.items())[:25]:
                try:
                    locator = page.locator(str(selector)).first
                    extracted[str(name)] = (await locator.inner_text(timeout=3000)).strip()
                except Exception:
                    extracted[str(name)] = None
            return {"status": "success", "data": extracted, "url": page.url}

        elif func_name == "upload_file":
            node_id = str(func_args.get("node_id") or "")
            file_path = str(func_args.get("file_path") or "")
            confirmation_id = str(func_args.get("confirmation_id") or "")
            approved = task.params.get("_approved_confirmation") if task and isinstance(task.params, dict) else None
            if not approved or confirmation_id != approved.get("id"):
                return {"status": "error", "error_code": "confirmation_required", "error": "Uploading data requires user confirmation."}
            from pathlib import Path
            from config.settings import settings

            safe_root = settings.jail_root.resolve()
            resolved = Path(file_path).resolve()
            if not resolved.is_relative_to(safe_root) or not resolved.is_file():
                return {"status": "error", "error_code": "invalid_upload_path", "error": "Upload path is outside the workspace."}
            await page.locator(f"[data-ax-node-id='{node_id}']").set_input_files(str(resolved), timeout=5000)
            task.params.pop("_approved_confirmation", None)
            return {"status": "success", "filename": resolved.name, "snapshot_revision": _advance_revision(page)}

        elif func_name == "download_file":
            node_id = str(func_args.get("node_id") or "")
            if not node_id:
                return {"error": "Missing 'node_id' argument."}
            async with page.expect_download(timeout=15000) as download_info:
                await page.locator(f"[data-ax-node-id='{node_id}']").click(timeout=3000)
            download = await download_info.value
            return {
                "status": "success",
                "filename": download.suggested_filename,
                "url": download.url,
                "snapshot_revision": _advance_revision(page),
            }
            
        elif func_name == "visual_check":
            goal = str(func_args.get("goal") or "").strip()
            if not goal:
                return {"status": "error", "error_code": "missing_goal", "error": "Missing 'goal' argument."}
            logger.info("Tool: Running ephemeral visual check")
            try:
                await page.wait_for_load_state("domcontentloaded", timeout=5000)
            except Exception:
                pass
            try:
                await page.evaluate(
                    "async () => { if (document.fonts?.ready) await document.fonts.ready; "
                    "await new Promise(requestAnimationFrame); await new Promise(requestAnimationFrame); }"
                )
            except Exception:
                pass
            screenshot_bytes = await page.screenshot(
                type="jpeg",
                quality=75,
                full_page=False,
                animations="disabled",
            )
            try:
                report = await vision_subagent.analyze([screenshot_bytes], goal)
            except VisionCheckError as exc:
                return {
                    "status": "error",
                    "error_code": "visual_check_failed",
                    "error": str(exc),
                    "retryable": exc.retryable,
                }
            report["snapshot_revision"] = _page_revision(page)
            return {"status": "success", "visual_check": report}

        elif func_name == "read_dom":
            logger.info("Tool: Reading compressed DOM")
            from browser.accessibility import extract_dom_text
            dom_text = await extract_dom_text(page)
            return {
                "status": "success",
                "dom_text": dom_text,
                "snapshot_revision": _advance_revision(page),
                "url": page.url,
                "title": await page.title(),
            }

        elif func_name == "read_text_content":
            logger.info("Tool: Reading full text content")
            try:
                # Playwright's innerText extracts human-readable text
                text = await page.evaluate("document.body.innerText")
                # Truncate to avoid blowing up the context completely (32000 chars is ~8k tokens)
                text = (text or "").strip()[:32000]
                return {
                    "status": "success",
                    "text_content": text,
                    "snapshot_revision": _page_revision(page),
                }
            except Exception as e:
                return {"error": f"Error reading text: {e}"}

        elif func_name == "find_element":
            query = func_args.get("query", "")
            if not query:
                return {"error": "Missing 'query' argument."}
            logger.info(f"Tool: Finding element for query '{query}'")
            import json
            # Escaping for JS
            q_escaped = query.replace("'", "\\'").lower()
            js = f'''
            (function(){{
                var q = '{q_escaped}';
                var els = Array.from(document.querySelectorAll("a,button,input,textarea,select,[role=button],[role=checkbox],[tabindex]"));
                window.__gravityAxNextId = window.__gravityAxNextId || 0;
                els.forEach((el) => {{
                    if (!el.getAttribute('data-ax-node-id')) {{
                        el.setAttribute('data-ax-node-id', 'ax-' + window.__gravityAxNextId++);
                    }}
                }});
                var found = els.filter(el => {{
                    var r = el.getBoundingClientRect();
                    var style = getComputedStyle(el);
                    var text = el.getAttribute('aria-label') || el.name || el.textContent || el.value || el.placeholder || "";
                    return style.display !== 'none' && style.visibility !== 'hidden' &&
                        r.width > 0 && r.height > 0 && text.toLowerCase().includes(q);
                }});
                return JSON.stringify(found.slice(0, 5).map(el => {{
                    var r = el.getBoundingClientRect();
                    var text = el.getAttribute('aria-label') || el.name || el.textContent || el.value || el.placeholder || "";
                    return {{
                        node_id: el.getAttribute('data-ax-node-id'),
                        tag: el.tagName,
                        text: text.trim().substring(0, 60),
                        x: Math.round(r.x + r.width/2),
                        y: Math.round(r.y + r.height/2)
                    }};
                }}));
            }})()
            '''
            try:
                result_json = await page.evaluate(js)
                elements = json.loads(result_json) if result_json else []
                return {"status": "success", "elements": elements}
            except Exception as e:
                return {"error": f"Error finding element: {e}"}
            
        elif func_name == "generate_image_and_deliver":
            prompt = func_args.get("prompt", "")
            service = func_args.get("service", "felo")
            if not prompt:
                return {"error": "Missing 'prompt' argument."}
            
            from workflows.account_manager import account_manager
            from workflows.ephemeral_media import ephemeral_media_store
            
            logger.info(f"Tool: Starte Bilderzeugung via {service} für '{prompt}'...")
            result = await account_manager.execute_task_with_auto_rotation(service, page, prompt)
            
            if result.success and result.media_bytes:
                media_id = ephemeral_media_store.put(
                    media_bytes=result.media_bytes,
                    media_type=result.media_type,
                    filename=result.filename,
                    ttl_seconds=900,  # 15 Minuten flüchtig im RAM
                )
                media_signal = f"[EPHEMERAL_IMAGE|{media_id}|{result.filename}|{prompt}]"
                return {
                    "status": "success",
                    "media_id": media_id,
                    "signal": media_signal,
                    "message": f"Bild erfolgreich generiert! Sende folgendes Signal an den Nutzer:\n\n{media_signal}\n\n{result.text_response}"
                }
            else:
                return {
                    "status": "error",
                    "error": result.error_message or "Bilderzeugung fehlgeschlagen."
                }

        else:
            return {"error": f"Unknown tool: {func_name}"}
            
    except Exception as e:
        logger.error(f"Error executing tool {func_name}: {e}")
        return {"error": str(e)}


async def execute_tool(
    func_name: str,
    func_args: Dict[str, Any],
    page: Page,
    task=None,
) -> Dict[str, Any]:
    """Execute a registered tool with a bounded timeout and stable outcome."""
    spec = TOOL_SPECS.get(func_name)
    if not spec:
        return {
            "status": "error",
            "outcome": "fatal",
            "error_code": "unknown_tool",
            "error": f"Unknown tool: {func_name}",
        }
    last_result: Dict[str, Any] = {}
    for attempt in range(1, spec.max_attempts + 1):
        try:
            root_page = getattr(page, "__dict__", {}).get("_gravity_root_page", page)
            active_page = getattr(root_page, "__dict__", {}).get(
                "_gravity_active_page",
                root_page,
            )
            result = await asyncio.wait_for(
                _execute_tool_impl(func_name, func_args, active_page, task=task),
                timeout=spec.timeout_seconds,
            )
        except asyncio.TimeoutError:
            result = {
                "status": "error",
                "error_code": "tool_timed_out",
                "error": f"{func_name} exceeded its {spec.timeout_seconds:.0f}s timeout.",
            }
        last_result = dict(result or {})
        is_error = bool(
            last_result.get("error")
            or last_result.get("status") == "error"
            or last_result.get("success") is False
        )
        if not is_error:
            last_result.setdefault("outcome", "success")
            last_result.setdefault("tool", func_name)
            last_result.setdefault("attempt", attempt)
            return last_result
        if not spec.idempotent or attempt >= spec.max_attempts:
            break
        await asyncio.sleep(0.15 * attempt)
    last_result.setdefault("status", "error")
    last_result.setdefault("outcome", "retryable_error" if spec.idempotent else "fatal")
    last_result.setdefault("error_code", "tool_failed")
    last_result.setdefault("tool", func_name)
    return last_result