Blame for engine/executor.py

"""
TokioAI Engine v3.0 -- Server-Side Tool Executor
Executes tools LOCALLY on the Pi without CLI round-trips.
This is what makes the Engine think like Opus: it can investigate autonomously.

Safety: only execute_local, read_file, search_files, diagnose are allowed server-side.
Write operations (write_file, edit_file) are ALWAYS delegated to the CLI for user approval.
Remote operations (execute_raspi, execute_gcp, execute_router) are also delegated.
"""

import os
import re
import subprocess
import time
import glob as glob_module
from typing import Dict, Tuple, Optional


# Tools that the Engine can execute directly (read-only / diagnostic)
SAFE_TOOLS = {
    "execute_local",
    "read_file",
    "search_files",
    "diagnose",
    "robot_vision",
}

# Tools that must be delegated to the CLI (writes, remote)
DELEGATE_TOOLS = {
    "write_file",
    "edit_file",
    "execute_raspi",
    "execute_gcp",
    "execute_router",
    "memory",
    "task",
}

# Commands that are NEVER allowed server-side (destructive)
BLOCKED_COMMANDS = [
    r'\brm\s+-rf\s+/',
    r'\bmkfs\b',
    r'\bdd\s+if=',
    r'\b(reboot|shutdown|poweroff|halt)\b',
    r'\bsudo\s+(rm|mkfs|dd|reboot|shutdown|poweroff|halt)\b',
    r'>\s*/dev/',
    r'\bchmod\s+-R\s+777\s+/',
    r'\bchown\s+-R.*\s+/',
]


def is_safe_tool(tool_name: str) -> bool:
    """Check if a tool can be executed server-side."""
    return tool_name in SAFE_TOOLS


def is_command_blocked(command: str) -> bool:
    """Check if a shell command is blocked for safety."""
    for pattern in BLOCKED_COMMANDS:
        if re.search(pattern, command, re.IGNORECASE):
            return True
    return False


def execute_tool(tool_name: str, arguments: Dict, timeout: int = 30) -> Tuple[bool, str]:
    """
    Execute a tool server-side.
    
    Returns:
        (success: bool, output: str)
        
    If the tool is not in SAFE_TOOLS, returns (False, "DELEGATE") to signal
    that this tool should be sent back to the CLI.
    """
    if tool_name not in SAFE_TOOLS:
        return False, "DELEGATE"
    
    try:
        if tool_name == "execute_local":
            return _execute_local(arguments, timeout)
        elif tool_name == "read_file":
            return _read_file(arguments)
        elif tool_name == "search_files":
            return _search_files(arguments)
        elif tool_name == "diagnose":
            return _diagnose(arguments)
        elif tool_name == "robot_vision":
            return _robot_vision(arguments)
        else:
            return False, f"Unknown tool: {tool_name}"
    except Exception as e:
        return False, f"Error executing {tool_name}: {str(e)}"


def _execute_local(arguments: Dict, default_timeout: int = 30) -> Tuple[bool, str]:
    """Execute a shell command locally."""
    command = arguments.get("command", "")
    if not command:
        return False, "No command specified"
    
    if is_command_blocked(command):
        return False, f"Command blocked for safety: {command}"
    
    timeout = min(arguments.get("timeout", default_timeout), 60)  # Cap at 60s for server-side
    
    try:
        result = subprocess.run(
            command,
            shell=True,
            capture_output=True,
            text=True,
            timeout=timeout,
            env={**os.environ, "TERM": "dumb"},  # prevent color codes
        )
        
        output = result.stdout
        if result.stderr:
            output += ("\n" if output else "") + result.stderr
        
        output = output.strip()
        
        if not output:
            if result.returncode == 0:
                output = "(no output)"
            else:
                output = f"(exit code {result.returncode}, no output)"
        
        # Truncate very long output to prevent context explosion
        if len(output) > 3000:
            output = output[:2500] + f"\n... [truncated, {len(output)} total chars]"
        
        success = (result.returncode == 0)
        return success, output
        
    except subprocess.TimeoutExpired:
        return False, f"Command timed out after {timeout}s"
    except Exception as e:
        return False, f"Execution error: {str(e)}"


def _read_file(arguments: Dict) -> Tuple[bool, str]:
    """Read a file from the local filesystem."""
    path = arguments.get("path", "")
    if not path:
        return False, "No path specified"
    
    # Only allow local paths (no raspi: or gcp: prefix)
    if ":" in path and not path.startswith("/"):
        return False, "DELEGATE"  # Remote file, delegate to CLI
    
    # Expand user home
    path = os.path.expanduser(path)
    
    if not os.path.exists(path):
        return False, f"No such file or directory: {path}"
    
    if os.path.isdir(path):
        return False, f"Is a directory: {path}"
    
    max_lines = arguments.get("lines", None)
    
    try:
        with open(path, "r", errors="replace") as f:
            if max_lines:
                content = "".join(f.readline() for _ in range(max_lines))
            else:
                content = f.read()
        
        # Truncate
        if len(content) > 5000:
            content = content[:4500] + f"\n... [truncated, {len(content)} total chars]"
        
        return True, content.strip() if content.strip() else "(empty file)"
        
    except PermissionError:
        return False, f"Permission denied: {path}"
    except Exception as e:
        return False, f"Error reading {path}: {str(e)}"


def _search_files(arguments: Dict) -> Tuple[bool, str]:
    """Search for patterns in files (like grep -rn)."""
    pattern = arguments.get("pattern", "")
    if not pattern:
        return False, "No pattern specified"
    
    search_path = arguments.get("path", ".")
    
    # Only local paths
    if ":" in search_path and not search_path.startswith("/"):
        return False, "DELEGATE"
    
    search_path = os.path.expanduser(search_path)
    file_glob = arguments.get("glob", "*")
    
    try:
        # Use grep for efficiency
        cmd = f"grep -rn --include='{file_glob}' '{pattern}' {search_path} 2>/dev/null | head -50"
        result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=15)
        output = result.stdout.strip()
        
        if not output:
            return True, f"No matches found for '{pattern}' in {search_path}"
        
        if len(output) > 3000:
            output = output[:2500] + "\n... [truncated]"
        
        return True, output
        
    except Exception as e:
        return False, f"Search error: {str(e)}"


def _diagnose(arguments: Dict) -> Tuple[bool, str]:
    """Run basic diagnostics on the local system."""
    target = arguments.get("target", "local")
    
    if target != "local":
        return False, "DELEGATE"  # Remote diagnostics delegate to CLI
    
    checks = []
    
    # CPU load
    try:
        with open("/proc/loadavg", "r") as f:
            load = f.read().strip()
        checks.append(f"Load: {load}")
    except:
        pass
    
    # Memory
    try:
        result = subprocess.run("free -h | head -3", shell=True, capture_output=True, text=True, timeout=5)
        checks.append(f"Memory:\n{result.stdout.strip()}")
    except:
        pass
    
    # Disk
    try:
        result = subprocess.run("df -h / | tail -1", shell=True, capture_output=True, text=True, timeout=5)
        checks.append(f"Disk: {result.stdout.strip()}")
    except:
        pass
    
    # Docker
    try:
        result = subprocess.run("docker ps --format 'table {{.Names}}\t{{.Status}}' 2>/dev/null | head -10", shell=True, capture_output=True, text=True, timeout=5)
        if result.returncode == 0 and result.stdout.strip():
            checks.append(f"Docker:\n{result.stdout.strip()}")
    except:
        pass
    
    # Key services
    try:
        result = subprocess.run("systemctl list-units --type=service --state=running --no-pager | head -20", shell=True, capture_output=True, text=True, timeout=5)
        checks.append(f"Services:\n{result.stdout.strip()}")
    except:
        pass
    
    return True, "\n\n".join(checks) if checks else "Could not gather diagnostics"


def _robot_vision(arguments: Dict) -> Tuple[bool, str]:
    """Capture a snapshot from the robot's camera and analyze with moondream vision model."""
    import requests
    import base64
    import time

    ROBOT_PROXY = os.environ.get("ROBOT_PROXY_URL", "http://localhost:5002")
    OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://localhost:8200")
    VISION_MODEL = os.environ.get("VISION_MODEL", "moondream")

    question = arguments.get("question", "")
    if not question:
        question = "Describe what you see in this image in detail. What objects, people, furniture, or environment is visible? Be specific."

    # 1. Capture snapshot from robot camera
    t0 = time.time()
    try:
        resp = requests.get(f"{ROBOT_PROXY}/snapshot", timeout=10)
        resp.raise_for_status()
        img_data = resp.content
    except requests.ConnectionError:
        return False, "Robot camera unavailable -- cannot connect to robot proxy (localhost:5002). Is picar-forward.service running?"
    except Exception as e:
        return False, f"Error capturing snapshot from robot camera: {e}"

    t_capture = time.time() - t0

    if len(img_data) < 500:
        return False, f"Robot camera returned suspiciously small image ({len(img_data)} bytes). Camera may be offline."

    # 2. Encode to base64
    img_b64 = base64.b64encode(img_data).decode('utf-8')

    # 3. Analyze with vision model
    t1 = time.time()
    try:
        resp = requests.post(
            f"{OLLAMA_URL}/api/generate",
            json={
                "model": VISION_MODEL,
                "prompt": question,
                "images": [img_b64],
                "stream": False,
            },
            timeout=60,
        )
        resp.raise_for_status()
        result = resp.json()
    except requests.ConnectionError:
        return False, "Vision model unavailable -- cannot connect to Ollama (localhost:8200). Is tokioai-coder-tunnel.service running?"
    except Exception as e:
        return False, f"Error analyzing image with vision model: {e}"

    t_analysis = time.time() - t1

    description = result.get("response", "").strip()
    if not description:
        return False, "Vision model returned empty response."

    tokens = result.get("eval_count", 0)

    output = (
        f"[Robot Vision -- moondream]\n"
        f"Image: {len(img_data):,} bytes | Capture: {t_capture:.1f}s | Analysis: {t_analysis:.1f}s | Tokens: {tokens}\n"
        f"Question: {question}\n"
        f"Answer: {description}"
    )

    return True, output