tools/real_executor.py

"""
REAL EXECUTION ENGINE — God Mode+ v4.0
Devin/Manus-style autonomous execution with:
- Real terminal execution with streaming
- File system control (read/write/patch/delete/move/search/tree)
- GitHub autonomy (clone/commit/push/PR/branch)
- Self-repair loop (error analysis + auto-patch + retry)
- Deployment automation (Vercel/HuggingFace)
- Tool call router
"""

import asyncio
import json
import os
import re
import subprocess
import tempfile
import time
import uuid
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

import httpx
import structlog

log = structlog.get_logger()

WORKSPACE = os.environ.get("WORKSPACE_DIR", "/tmp/god_workspace")
GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN", "")
HF_TOKEN = os.environ.get("HF_TOKEN", "")
VERCEL_TOKEN = os.environ.get("VERCEL_TOKEN", "")

os.makedirs(WORKSPACE, exist_ok=True)

# ─── BLOCKED COMMANDS (Security) ────────────────────────────────────────────
BLOCKED_PATTERNS = [
    r"rm\s+-rf\s+/[^t]",  # rm -rf / but allow /tmp
    r":\(\)\s*\{",         # fork bomb
    r"mkfs\.",             # format filesystem
    r"dd\s+if=/dev/",     # disk destruction
    r"shutdown",
    r"reboot",
    r"halt",
    r"poweroff",
    r"chmod\s+777\s+/",
    r"wget.*\|\s*bash",
    r"curl.*\|\s*bash",
]

def is_blocked(cmd: str) -> bool:
    for pattern in BLOCKED_PATTERNS:
        if re.search(pattern, cmd, re.IGNORECASE):
            return True
    return False


# ═══════════════════════════════════════════════════════════════════
# TERMINAL EXECUTION ENGINE
# ═══════════════════════════════════════════════════════════════════

class TerminalEngine:
    """Real shell execution with streaming output."""

    def __init__(self, ws_manager=None):
        self.ws = ws_manager
        self._sessions: Dict[str, str] = {}  # session_id -> cwd

    def get_cwd(self, session_id: str) -> str:
        return self._sessions.get(session_id, WORKSPACE)

    def set_cwd(self, session_id: str, cwd: str):
        if os.path.isdir(cwd):
            self._sessions[session_id] = cwd

    async def execute(
        self,
        command: str,
        session_id: str = "",
        task_id: str = "",
        cwd: Optional[str] = None,
        timeout: int = 60,
        env_vars: Dict[str, str] = {},
    ) -> Dict[str, Any]:
        """Execute command with real streaming output."""

        if is_blocked(command):
            return {
                "success": False,
                "output": f"❌ BLOCKED: Dangerous command rejected: {command[:80]}",
                "exit_code": -1,
                "command": command,
            }

        work_dir = cwd or self.get_cwd(session_id)
        os.makedirs(work_dir, exist_ok=True)

        # Handle cd specially
        if command.strip().startswith("cd "):
            new_dir = command.strip()[3:].strip()
            if not os.path.isabs(new_dir):
                new_dir = os.path.join(work_dir, new_dir)
            new_dir = os.path.normpath(new_dir)
            if os.path.isdir(new_dir):
                self.set_cwd(session_id, new_dir)
                return {"success": True, "output": f"📂 Changed to: {new_dir}", "exit_code": 0, "command": command, "cwd": new_dir}
            else:
                return {"success": False, "output": f"❌ Directory not found: {new_dir}", "exit_code": 1, "command": command}

        # Build environment
        env = {**os.environ, **env_vars}

        # Emit start event
        if self.ws and task_id:
            await self.ws.emit(task_id, "terminal_start", {
                "command": command[:200],
                "cwd": work_dir,
                "session_id": session_id,
            }, session_id=session_id)

        output_lines = []
        exit_code = 0

        try:
            proc = await asyncio.create_subprocess_shell(
                command,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.STDOUT,
                cwd=work_dir,
                env=env,
            )

            # Stream output line by line
            async def read_stream():
                while True:
                    try:
                        line = await asyncio.wait_for(proc.stdout.readline(), timeout=timeout)
                        if not line:
                            break
                        decoded = line.decode("utf-8", errors="replace").rstrip()
                        output_lines.append(decoded)

                        # Stream to websocket
                        if self.ws and (task_id or session_id):
                            await self.ws.emit(
                                task_id or "logs",
                                "terminal_output",
                                {"line": decoded, "command": command[:80]},
                                session_id=session_id,
                            )
                    except asyncio.TimeoutError:
                        proc.kill()
                        output_lines.append(f"\n⏱️ Command timed out after {timeout}s")
                        break

            await read_stream()
            try:
                exit_code = await asyncio.wait_for(proc.wait(), timeout=5)
            except asyncio.TimeoutError:
                proc.kill()
                exit_code = -1

        except Exception as e:
            output_lines.append(f"❌ Execution error: {str(e)}")
            exit_code = -1

        full_output = "\n".join(output_lines)

        # Emit completion
        if self.ws and task_id:
            await self.ws.emit(task_id, "terminal_done", {
                "command": command[:80],
                "exit_code": exit_code,
                "success": exit_code == 0,
                "lines": len(output_lines),
            }, session_id=session_id)

        return {
            "success": exit_code == 0,
            "output": full_output[:8000],
            "exit_code": exit_code,
            "command": command,
            "cwd": work_dir,
        }

    async def run_sequence(
        self,
        commands: List[str],
        session_id: str = "",
        task_id: str = "",
        stop_on_error: bool = True,
    ) -> List[Dict]:
        """Run a sequence of commands, streaming each."""
        results = []
        for cmd in commands:
            result = await self.execute(cmd, session_id=session_id, task_id=task_id)
            results.append(result)
            if not result["success"] and stop_on_error:
                break
        return results


# ═══════════════════════════════════════════════════════════════════
# FILESYSTEM CONTROL ENGINE
# ═══════════════════════════════════════════════════════════════════

class FilesystemEngine:
    """Full file system control."""

    def __init__(self, ws_manager=None):
        self.ws = ws_manager

    def _resolve(self, path: str, base: str = WORKSPACE) -> str:
        if os.path.isabs(path):
            resolved = os.path.normpath(path)
        else:
            resolved = os.path.normpath(os.path.join(base, path))
        return resolved

    async def read_file(self, path: str, session_id: str = "", task_id: str = "") -> Dict:
        fpath = self._resolve(path)
        try:
            with open(fpath, "r", encoding="utf-8", errors="replace") as f:
                content = f.read()
            lines = content.split("\n")
            result = {
                "success": True,
                "path": fpath,
                "content": content,
                "lines": len(lines),
                "size": len(content),
                "language": self._detect_language(fpath),
            }
            if self.ws and task_id:
                await self.ws.emit(task_id, "file_read", {"path": fpath, "lines": len(lines)}, session_id=session_id)
            return result
        except FileNotFoundError:
            return {"success": False, "error": f"File not found: {path}"}
        except Exception as e:
            return {"success": False, "error": str(e)}

    async def write_file(self, path: str, content: str, session_id: str = "", task_id: str = "") -> Dict:
        fpath = self._resolve(path)
        os.makedirs(os.path.dirname(fpath), exist_ok=True)
        try:
            with open(fpath, "w", encoding="utf-8") as f:
                f.write(content)
            lines = len(content.split("\n"))
            if self.ws and task_id:
                await self.ws.emit(task_id, "file_written", {
                    "path": fpath,
                    "filename": os.path.basename(fpath),
                    "size": len(content),
                    "lines": lines,
                    "language": self._detect_language(fpath),
                }, session_id=session_id)
            return {"success": True, "path": fpath, "size": len(content), "lines": lines}
        except Exception as e:
            return {"success": False, "error": str(e)}

    async def patch_file(self, path: str, old_content: str, new_content: str, session_id: str = "", task_id: str = "") -> Dict:
        """Patch a specific section of a file."""
        fpath = self._resolve(path)
        try:
            with open(fpath, "r", encoding="utf-8") as f:
                current = f.read()
            if old_content not in current:
                return {"success": False, "error": "Patch target not found in file"}
            patched = current.replace(old_content, new_content, 1)
            with open(fpath, "w", encoding="utf-8") as f:
                f.write(patched)

            # Emit diff event
            if self.ws and task_id:
                await self.ws.emit(task_id, "file_patched", {
                    "path": fpath,
                    "filename": os.path.basename(fpath),
                    "old_snippet": old_content[:100],
                    "new_snippet": new_content[:100],
                }, session_id=session_id)

            return {
                "success": True,
                "path": fpath,
                "diff": {
                    "removed": old_content.split("\n")[:5],
                    "added": new_content.split("\n")[:5],
                }
            }
        except Exception as e:
            return {"success": False, "error": str(e)}

    async def delete_file(self, path: str, session_id: str = "", task_id: str = "") -> Dict:
        fpath = self._resolve(path)
        try:
            os.remove(fpath)
            if self.ws and task_id:
                await self.ws.emit(task_id, "file_deleted", {"path": fpath}, session_id=session_id)
            return {"success": True, "path": fpath}
        except Exception as e:
            return {"success": False, "error": str(e)}

    async def move_file(self, src: str, dst: str, session_id: str = "", task_id: str = "") -> Dict:
        import shutil
        src_path = self._resolve(src)
        dst_path = self._resolve(dst)
        try:
            os.makedirs(os.path.dirname(dst_path), exist_ok=True)
            shutil.move(src_path, dst_path)
            if self.ws and task_id:
                await self.ws.emit(task_id, "file_moved", {"from": src_path, "to": dst_path}, session_id=session_id)
            return {"success": True, "from": src_path, "to": dst_path}
        except Exception as e:
            return {"success": False, "error": str(e)}

    async def search_files(self, query: str, path: str = "", file_pattern: str = "*") -> Dict:
        """Search file contents with grep."""
        search_path = self._resolve(path) if path else WORKSPACE
        try:
            result = subprocess.run(
                ["grep", "-r", "-l", "--include", file_pattern, query, search_path],
                capture_output=True, text=True, timeout=10
            )
            files = [f for f in result.stdout.strip().split("\n") if f]
            return {"success": True, "files": files, "count": len(files), "query": query}
        except Exception as e:
            return {"success": False, "error": str(e), "files": []}

    async def tree(self, path: str = "", max_depth: int = 4) -> Dict:
        """Get file tree of workspace."""
        root = self._resolve(path) if path else WORKSPACE
        try:
            result = subprocess.run(
                ["find", root, "-maxdepth", str(max_depth),
                 "!", "-path", "*/node_modules/*",
                 "!", "-path", "*/__pycache__/*",
                 "!", "-path", "*/.git/*",
                 "!", "-path", "*/.next/*",
                 "-type", "f"],
                capture_output=True, text=True, timeout=10
            )
            files = [os.path.relpath(f, root) for f in result.stdout.strip().split("\n") if f]
            return {"success": True, "root": root, "files": files[:200], "count": len(files)}
        except Exception as e:
            return {"success": False, "error": str(e), "files": []}

    async def list_dir(self, path: str = "") -> Dict:
        target = self._resolve(path) if path else WORKSPACE
        try:
            entries = []
            for item in sorted(os.listdir(target)):
                full = os.path.join(target, item)
                entries.append({
                    "name": item,
                    "type": "dir" if os.path.isdir(full) else "file",
                    "size": os.path.getsize(full) if os.path.isfile(full) else 0,
                })
            return {"success": True, "path": target, "entries": entries}
        except Exception as e:
            return {"success": False, "error": str(e), "entries": []}

    def _detect_language(self, path: str) -> str:
        ext_map = {
            ".py": "python", ".ts": "typescript", ".tsx": "tsx",
            ".js": "javascript", ".jsx": "jsx", ".go": "go",
            ".rs": "rust", ".java": "java", ".cpp": "cpp",
            ".c": "c", ".sh": "bash", ".yaml": "yaml",
            ".yml": "yaml", ".json": "json", ".md": "markdown",
            ".html": "html", ".css": "css", ".sql": "sql",
            ".dockerfile": "dockerfile", ".toml": "toml",
        }
        ext = Path(path).suffix.lower()
        return ext_map.get(ext, "text")


# ═══════════════════════════════════════════════════════════════════
# GITHUB AUTONOMY ENGINE
# ═══════════════════════════════════════════════════════════════════

class GitHubEngine:
    """Full GitHub autonomy - clone, commit, push, PR, branch."""

    def __init__(self, token: str = "", ws_manager=None):
        self.token = token or GITHUB_TOKEN
        self.ws = ws_manager
        self.base_url = "https://api.github.com"
        self.headers = {
            "Authorization": f"token {self.token}",
            "Accept": "application/vnd.github.v3+json",
        }

    async def _api(self, method: str, path: str, data: Dict = {}) -> Dict:
        async with httpx.AsyncClient(timeout=30) as client:
            url = f"{self.base_url}{path}"
            if method == "GET":
                r = await client.get(url, headers=self.headers)
            elif method == "POST":
                r = await client.post(url, json=data, headers=self.headers)
            elif method == "PATCH":
                r = await client.patch(url, json=data, headers=self.headers)
            elif method == "PUT":
                r = await client.put(url, json=data, headers=self.headers)
            elif method == "DELETE":
                r = await client.delete(url, headers=self.headers)
            else:
                return {"error": f"Unknown method: {method}"}
            try:
                return r.json()
            except Exception:
                return {"error": r.text, "status": r.status_code}

    async def get_user(self) -> Dict:
        return await self._api("GET", "/user")

    async def clone_repo(
        self,
        repo_url: str,
        target_dir: str = "",
        session_id: str = "",
        task_id: str = "",
    ) -> Dict:
        """Clone a GitHub repository into workspace."""
        if not target_dir:
            repo_name = repo_url.rstrip("/").split("/")[-1].replace(".git", "")
            target_dir = os.path.join(WORKSPACE, repo_name)

        # Inject token into URL if available
        if self.token and "github.com" in repo_url:
            repo_url = repo_url.replace("https://", f"https://x-access-token:{self.token}@")

        if self.ws and task_id:
            await self.ws.emit(task_id, "github_op", {
                "op": "clone",
                "repo": repo_url.split("@")[-1] if "@" in repo_url else repo_url,
                "target": target_dir,
            }, session_id=session_id)

        try:
            proc = await asyncio.create_subprocess_exec(
                "git", "clone", "--depth", "50", repo_url, target_dir,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
            stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=120)
            success = proc.returncode == 0
            return {
                "success": success,
                "path": target_dir,
                "output": (stdout or stderr).decode("utf-8", errors="replace")[:1000],
            }
        except Exception as e:
            return {"success": False, "error": str(e)}

    async def create_repo(
        self,
        name: str,
        description: str = "",
        private: bool = False,
        auto_init: bool = True,
        session_id: str = "",
        task_id: str = "",
    ) -> Dict:
        """Create a new GitHub repository."""
        data = {
            "name": name,
            "description": description,
            "private": private,
            "auto_init": auto_init,
        }
        if self.ws and task_id:
            await self.ws.emit(task_id, "github_op", {"op": "create_repo", "name": name}, session_id=session_id)

        result = await self._api("POST", "/user/repos", data)
        return {
            "success": "html_url" in result,
            "url": result.get("html_url", ""),
            "clone_url": result.get("clone_url", ""),
            "ssh_url": result.get("ssh_url", ""),
            "name": result.get("name", name),
            "error": result.get("message", ""),
        }

    async def commit_and_push(
        self,
        repo_path: str,
        message: str,
        branch: str = "main",
        session_id: str = "",
        task_id: str = "",
    ) -> Dict:
        """Stage all changes, commit, and push."""
        commands = [
            f"git -C {repo_path} add -A",
            f'git -C {repo_path} commit -m "{message}" --allow-empty',
            f"git -C {repo_path} push origin {branch} --force-with-lease 2>&1 || git -C {repo_path} push origin {branch}",
        ]

        if self.ws and task_id:
            await self.ws.emit(task_id, "github_op", {
                "op": "commit_push",
                "message": message,
                "branch": branch,
            }, session_id=session_id)

        results = []
        for cmd in commands:
            proc = await asyncio.create_subprocess_shell(
                cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                env={**os.environ, "GIT_TERMINAL_PROMPT": "0"},
            )
            stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=60)
            out = (stdout or b"").decode("utf-8", errors="replace")
            err = (stderr or b"").decode("utf-8", errors="replace")
            results.append({
                "cmd": cmd.split("'")[0][:60],
                "success": proc.returncode == 0,
                "output": (out + err)[:500],
            })

        return {
            "success": all(r["success"] for r in results),
            "steps": results,
            "message": message,
            "branch": branch,
        }

    async def create_branch(
        self,
        repo_path: str,
        branch_name: str,
        from_branch: str = "main",
        session_id: str = "",
        task_id: str = "",
    ) -> Dict:
        cmd = f"git -C {repo_path} checkout -b {branch_name} origin/{from_branch} 2>/dev/null || git -C {repo_path} checkout -b {branch_name}"
        proc = await asyncio.create_subprocess_shell(cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=30)
        return {"success": proc.returncode == 0, "branch": branch_name}

    async def create_pr(
        self,
        owner: str,
        repo: str,
        title: str,
        body: str,
        head: str,
        base: str = "main",
        session_id: str = "",
        task_id: str = "",
    ) -> Dict:
        """Create a Pull Request."""
        data = {"title": title, "body": body, "head": head, "base": base}
        if self.ws and task_id:
            await self.ws.emit(task_id, "github_op", {"op": "create_pr", "title": title}, session_id=session_id)
        result = await self._api("POST", f"/repos/{owner}/{repo}/pulls", data)
        return {
            "success": "html_url" in result,
            "url": result.get("html_url", ""),
            "number": result.get("number"),
            "error": result.get("message", ""),
        }

    async def get_repo_contents(self, owner: str, repo: str, path: str = "") -> Dict:
        result = await self._api("GET", f"/repos/{owner}/{repo}/contents/{path}")
        return result

    async def read_issues(self, owner: str, repo: str, state: str = "open") -> List[Dict]:
        result = await self._api("GET", f"/repos/{owner}/{repo}/issues?state={state}&per_page=10")
        if isinstance(result, list):
            return [{"number": i.get("number"), "title": i.get("title"), "body": (i.get("body") or "")[:200]} for i in result]
        return []


# ═══════════════════════════════════════════════════════════════════
# SELF-REPAIR LOOP ENGINE
# ═══════════════════════════════════════════════════════════════════

class SelfRepairEngine:
    """
    Devin-style self-repair loop:
    Run → See error → Analyze → Patch → Retry → Pass
    """

    def __init__(self, terminal: TerminalEngine, filesystem: FilesystemEngine, ai_router=None, ws_manager=None):
        self.terminal = terminal
        self.filesystem = filesystem
        self.ai_router = ai_router
        self.ws = ws_manager

    async def run_with_repair(
        self,
        command: str,
        related_files: List[str] = [],
        session_id: str = "",
        task_id: str = "",
        max_retries: int = 3,
    ) -> Dict:
        """Run command, auto-repair on failure, retry."""

        attempt = 0
        last_result = None

        while attempt < max_retries:
            attempt += 1

            if self.ws and task_id and attempt > 1:
                await self.ws.emit(task_id, "retry_attempt", {
                    "attempt": attempt,
                    "max_retries": max_retries,
                    "command": command[:80],
                }, session_id=session_id)

            result = await self.terminal.execute(command, session_id=session_id, task_id=task_id)
            last_result = result

            if result["success"]:
                if attempt > 1:
                    if self.ws and task_id:
                        await self.ws.emit(task_id, "self_heal_success", {
                            "attempts": attempt,
                            "command": command[:80],
                        }, session_id=session_id)
                return {**result, "attempts": attempt, "self_healed": attempt > 1}

            # Failed — analyze error and attempt repair
            if self.ws and task_id:
                await self.ws.emit(task_id, "self_heal_attempt", {
                    "attempt": attempt,
                    "error_snippet": result["output"][-500:],
                    "command": command[:80],
                }, session_id=session_id)

            if not self.ai_router:
                break

            # Ask AI to analyze error and suggest fix
            patch = await self._analyze_and_repair(
                command=command,
                error_output=result["output"],
                related_files=related_files,
                session_id=session_id,
                task_id=task_id,
            )

            if patch and patch.get("fixed"):
                # Apply patches to files
                for file_patch in patch.get("file_patches", []):
                    fpath = file_patch.get("file", "")
                    old = file_patch.get("old", "")
                    new = file_patch.get("new", "")
                    if fpath and old and new:
                        await self.filesystem.patch_file(fpath, old, new, session_id=session_id, task_id=task_id)

                # Update command if needed
                if patch.get("new_command"):
                    command = patch["new_command"]

                # Install missing packages if needed
                if patch.get("install_command"):
                    await self.terminal.execute(patch["install_command"], session_id=session_id, task_id=task_id)

            await asyncio.sleep(1)

        # All retries exhausted
        if self.ws and task_id:
            await self.ws.emit(task_id, "self_heal_failed", {
                "attempts": attempt,
                "final_error": (last_result or {}).get("output", "")[-300:],
            }, session_id=session_id)

        return {**(last_result or {}), "attempts": attempt, "self_healed": False}

    async def _analyze_and_repair(
        self,
        command: str,
        error_output: str,
        related_files: List[str],
        session_id: str,
        task_id: str,
    ) -> Dict:
        """Use AI to analyze error and produce repair instructions."""

        # Read related file contents for context
        file_context = ""
        for fp in related_files[:3]:
            try:
                full_path = os.path.join(WORKSPACE, fp)
                if os.path.exists(full_path):
                    with open(full_path, "r", errors="replace") as f:
                        content = f.read()[:1000]
                    file_context += f"\n\nFile: {fp}\n```\n{content}\n```"
            except Exception:
                pass

        messages = [
            {
                "role": "system",
                "content": (
                    "You are an expert debugger. Analyze the error and provide a precise repair.\n"
                    "Respond ONLY with valid JSON.\n"
                    "Format:\n"
                    '{"fixed": true/false, '
                    '"analysis": "brief error explanation", '
                    '"fix_description": "what to fix", '
                    '"install_command": "pip install X or npm install X if missing package, else null", '
                    '"new_command": "corrected command if needed, else null", '
                    '"file_patches": [{"file": "filename", "old": "exact old text", "new": "exact new text"}]}'
                ),
            },
            {
                "role": "user",
                "content": (
                    f"Command that failed:\n```\n{command}\n```\n\n"
                    f"Error output:\n```\n{error_output[-2000:]}\n```"
                    f"{file_context}"
                ),
            },
        ]

        try:
            raw = await self.ai_router.complete(messages, temperature=0.1, max_tokens=1000)
            start = raw.find("{")
            end = raw.rfind("}") + 1
            if start >= 0 and end > start:
                return json.loads(raw[start:end])
        except Exception as e:
            log.warning("Self-repair analysis failed", error=str(e))

        return {"fixed": False}


# ═══════════════════════════════════════════════════════════════════
# DEPLOYMENT ENGINE
# ═══════════════════════════════════════════════════════════════════

class DeploymentEngine:
    """Deployment automation for Vercel + HuggingFace."""

    def __init__(self, ws_manager=None):
        self.ws = ws_manager
        self.vercel_token = VERCEL_TOKEN
        self.hf_token = HF_TOKEN

    async def deploy_vercel(
        self,
        project_dir: str,
        project_name: str,
        session_id: str = "",
        task_id: str = "",
    ) -> Dict:
        """Deploy a project to Vercel."""
        if self.ws and task_id:
            await self.ws.emit(task_id, "deploy_start", {
                "platform": "vercel",
                "project": project_name,
            }, session_id=session_id)

        # Use vercel CLI if available
        env = {**os.environ, "VERCEL_TOKEN": self.vercel_token}
        try:
            proc = await asyncio.create_subprocess_shell(
                f"cd {project_dir} && npx vercel --token={self.vercel_token} --yes --prod 2>&1",
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.STDOUT,
                env=env,
            )
            stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=300)
            output = stdout.decode("utf-8", errors="replace")

            # Extract URL from output
            url_match = re.search(r"https://[a-zA-Z0-9\-]+\.vercel\.app", output)
            url = url_match.group(0) if url_match else ""

            success = proc.returncode == 0 and bool(url)

            if self.ws and task_id:
                await self.ws.emit(task_id, "deploy_complete", {
                    "platform": "vercel",
                    "success": success,
                    "url": url,
                    "project": project_name,
                }, session_id=session_id)

            return {"success": success, "url": url, "output": output[-1000:], "platform": "vercel"}

        except Exception as e:
            return {"success": False, "error": str(e), "platform": "vercel"}

    async def deploy_hf_space(
        self,
        repo_path: str,
        space_name: str,
        space_type: str = "gradio",
        session_id: str = "",
        task_id: str = "",
    ) -> Dict:
        """Push to HuggingFace Space."""
        if self.ws and task_id:
            await self.ws.emit(task_id, "deploy_start", {
                "platform": "huggingface",
                "space": space_name,
            }, session_id=session_id)

        hf_url = f"https://huggingface.co/spaces/{space_name}"
        git_url = f"https://huggingface.co/spaces/{space_name}.git"

        # Inject HF token
        auth_url = git_url.replace("https://", f"https://user:{self.hf_token}@")

        try:
            cmds = [
                f"cd {repo_path} && git remote remove hf_space 2>/dev/null; git remote add hf_space {auth_url}",
                f"cd {repo_path} && git push hf_space main --force 2>&1",
            ]
            results = []
            for cmd in cmds:
                proc = await asyncio.create_subprocess_shell(cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT)
                stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=120)
                results.append({"cmd": cmd[:40], "success": proc.returncode == 0, "out": stdout.decode("utf-8", errors="replace")[:300]})

            success = results[-1]["success"]
            if self.ws and task_id:
                await self.ws.emit(task_id, "deploy_complete", {
                    "platform": "huggingface",
                    "success": success,
                    "url": hf_url,
                    "space": space_name,
                }, session_id=session_id)

            return {"success": success, "url": hf_url, "platform": "huggingface", "steps": results}

        except Exception as e:
            return {"success": False, "error": str(e), "platform": "huggingface"}


# ═══════════════════════════════════════════════════════════════════
# UNIFIED REAL TOOL ROUTER
# ═══════════════════════════════════════════════════════════════════

class RealToolRouter:
    """
    Routes tool calls to real implementations.
    This is the core of the autonomous execution system.
    """

    def __init__(self, ws_manager=None, ai_router=None):
        self.ws = ws_manager
        self.ai_router = ai_router
        self.terminal = TerminalEngine(ws_manager=ws_manager)
        self.filesystem = FilesystemEngine(ws_manager=ws_manager)
        self.github = GitHubEngine(ws_manager=ws_manager)
        self.repair = SelfRepairEngine(
            terminal=self.terminal,
            filesystem=self.filesystem,
            ai_router=ai_router,
            ws_manager=ws_manager,
        )
        self.deployment = DeploymentEngine(ws_manager=ws_manager)

    async def route(
        self,
        tool: str,
        params: Dict,
        session_id: str = "",
        task_id: str = "",
    ) -> Dict:
        """Route a tool call to the correct engine."""

        # Emit tool call event
        if self.ws and task_id:
            await self.ws.emit(task_id, "tool_called", {
                "tool": tool,
                "params_preview": str(params)[:100],
            }, session_id=session_id)

        handlers = {
            # Terminal
            "terminal.run": self._terminal_run,
            "terminal.sequence": self._terminal_sequence,
            "terminal.run_with_repair": self._terminal_repair,

            # Filesystem
            "fs.read": self._fs_read,
            "fs.write": self._fs_write,
            "fs.patch": self._fs_patch,
            "fs.delete": self._fs_delete,
            "fs.move": self._fs_move,
            "fs.search": self._fs_search,
            "fs.tree": self._fs_tree,
            "fs.list": self._fs_list,

            # GitHub
            "github.clone": self._github_clone,
            "github.create_repo": self._github_create_repo,
            "github.commit_push": self._github_commit_push,
            "github.create_branch": self._github_create_branch,
            "github.create_pr": self._github_create_pr,
            "github.read_issues": self._github_issues,

            # Deployment
            "deploy.vercel": self._deploy_vercel,
            "deploy.hf": self._deploy_hf,

            # Meta
            "workspace.info": self._workspace_info,
        }

        handler = handlers.get(tool)
        if not handler:
            return {"success": False, "error": f"Unknown tool: {tool}", "available": list(handlers.keys())}

        try:
            result = await handler(params, session_id=session_id, task_id=task_id)

            # Emit result event
            if self.ws and task_id:
                await self.ws.emit(task_id, "tool_result", {
                    "tool": tool,
                    "success": result.get("success", True),
                    "summary": str(result)[:150],
                }, session_id=session_id)

            return result
        except Exception as e:
            log.error("Tool execution failed", tool=tool, error=str(e))
            return {"success": False, "error": str(e), "tool": tool}

    # ─── Terminal Handlers ────────────────────────────────────────────────────
    async def _terminal_run(self, params: Dict, session_id: str, task_id: str) -> Dict:
        return await self.terminal.execute(
            params.get("command", ""),
            session_id=session_id,
            task_id=task_id,
            cwd=params.get("cwd"),
            timeout=params.get("timeout", 60),
        )

    async def _terminal_sequence(self, params: Dict, session_id: str, task_id: str) -> Dict:
        results = await self.terminal.run_sequence(
            params.get("commands", []),
            session_id=session_id,
            task_id=task_id,
            stop_on_error=params.get("stop_on_error", True),
        )
        return {"success": all(r["success"] for r in results), "steps": results}

    async def _terminal_repair(self, params: Dict, session_id: str, task_id: str) -> Dict:
        return await self.repair.run_with_repair(
            params.get("command", ""),
            related_files=params.get("related_files", []),
            session_id=session_id,
            task_id=task_id,
            max_retries=params.get("max_retries", 3),
        )

    # ─── Filesystem Handlers ──────────────────────────────────────────────────
    async def _fs_read(self, params: Dict, session_id: str, task_id: str) -> Dict:
        return await self.filesystem.read_file(params.get("path", ""), session_id=session_id, task_id=task_id)

    async def _fs_write(self, params: Dict, session_id: str, task_id: str) -> Dict:
        return await self.filesystem.write_file(params.get("path", ""), params.get("content", ""), session_id=session_id, task_id=task_id)

    async def _fs_patch(self, params: Dict, session_id: str, task_id: str) -> Dict:
        return await self.filesystem.patch_file(params.get("path", ""), params.get("old", ""), params.get("new", ""), session_id=session_id, task_id=task_id)

    async def _fs_delete(self, params: Dict, session_id: str, task_id: str) -> Dict:
        return await self.filesystem.delete_file(params.get("path", ""), session_id=session_id, task_id=task_id)

    async def _fs_move(self, params: Dict, session_id: str, task_id: str) -> Dict:
        return await self.filesystem.move_file(params.get("src", ""), params.get("dst", ""), session_id=session_id, task_id=task_id)

    async def _fs_search(self, params: Dict, session_id: str, task_id: str) -> Dict:
        return await self.filesystem.search_files(params.get("query", ""), params.get("path", ""), params.get("pattern", "*"))

    async def _fs_tree(self, params: Dict, session_id: str, task_id: str) -> Dict:
        return await self.filesystem.tree(params.get("path", ""), params.get("max_depth", 4))

    async def _fs_list(self, params: Dict, session_id: str, task_id: str) -> Dict:
        return await self.filesystem.list_dir(params.get("path", ""))

    # ─── GitHub Handlers ──────────────────────────────────────────────────────
    async def _github_clone(self, params: Dict, session_id: str, task_id: str) -> Dict:
        return await self.github.clone_repo(params.get("url", ""), params.get("target", ""), session_id=session_id, task_id=task_id)

    async def _github_create_repo(self, params: Dict, session_id: str, task_id: str) -> Dict:
        return await self.github.create_repo(params.get("name", ""), params.get("description", ""), params.get("private", False), session_id=session_id, task_id=task_id)

    async def _github_commit_push(self, params: Dict, session_id: str, task_id: str) -> Dict:
        return await self.github.commit_and_push(params.get("repo_path", WORKSPACE), params.get("message", "God Agent commit"), params.get("branch", "main"), session_id=session_id, task_id=task_id)

    async def _github_create_branch(self, params: Dict, session_id: str, task_id: str) -> Dict:
        return await self.github.create_branch(params.get("repo_path", WORKSPACE), params.get("branch", ""), session_id=session_id, task_id=task_id)

    async def _github_create_pr(self, params: Dict, session_id: str, task_id: str) -> Dict:
        return await self.github.create_pr(params.get("owner", ""), params.get("repo", ""), params.get("title", ""), params.get("body", ""), params.get("head", ""), session_id=session_id, task_id=task_id)

    async def _github_issues(self, params: Dict, session_id: str, task_id: str) -> Dict:
        issues = await self.github.read_issues(params.get("owner", ""), params.get("repo", ""))
        return {"success": True, "issues": issues}

    # ─── Deployment Handlers ──────────────────────────────────────────────────
    async def _deploy_vercel(self, params: Dict, session_id: str, task_id: str) -> Dict:
        return await self.deployment.deploy_vercel(params.get("dir", WORKSPACE), params.get("name", "god-agent-app"), session_id=session_id, task_id=task_id)

    async def _deploy_hf(self, params: Dict, session_id: str, task_id: str) -> Dict:
        return await self.deployment.deploy_hf_space(params.get("repo_path", WORKSPACE), params.get("space_name", ""), session_id=session_id, task_id=task_id)

    # ─── Meta ─────────────────────────────────────────────────────────────────
    async def _workspace_info(self, params: Dict, session_id: str, task_id: str) -> Dict:
        tree = await self.filesystem.tree()
        return {
            "success": True,
            "workspace": WORKSPACE,
            "files": tree.get("files", []),
            "file_count": tree.get("count", 0),
        }


# ─── Singleton instance ────────────────────────────────────────────────────────
_tool_router: Optional[RealToolRouter] = None

def get_tool_router(ws_manager=None, ai_router=None) -> RealToolRouter:
    global _tool_router
    if _tool_router is None:
        _tool_router = RealToolRouter(ws_manager=ws_manager, ai_router=ai_router)
    return _tool_router