Upload litehat/mcp_terminal.py

litehat/mcp_terminal.py

@@ -0,0 +1,329 @@
+"""
+LITEHAT MCP TERMINAL
+Native terminal environment via Model Context Protocol.
+
+Unlike Docker-only isolation, Litehat uses MCP to access:
+- Local filesystem with zero-copy operations
+- System shell with full process control
+- Browser automation via Playwright/Puppeteer
+- Package managers (npm, pip, cargo, etc.)
+- Git repositories and version control
+- Network access for API calls and webhooks
+
+This gives Litehat REAL hands in the system — not a sandbox.
+"""
+
+import os
+import sys
+import subprocess
+import json
+import shutil
+import tempfile
+import asyncio
+from pathlib import Path
+from typing import Optional, Dict, Any, List
+from dataclasses import dataclass, field
+
+
+@dataclass
+class TerminalResult:
+    """Result of a terminal command execution."""
+    command: str
+    stdout: str
+    stderr: str
+    exit_code: int
+    elapsed_ms: float
+    working_dir: str
+
+
+@dataclass
+class FileOperation:
+    """Record of a file operation for undo/rollback."""
+    path: str
+    operation: str  # "create", "modify", "delete"
+    backup_content: Optional[str] = None  # For rollback
+    timestamp: float = 0.0
+
+
+class MCPTerminal:
+    """
+    Model Context Protocol Terminal — Litehat's hands in the system.
+
+    Provides native access to:
+    - Filesystem (read/write/delete/traverse)
+    - Shell execution (blocking and streaming)
+    - Package management (npm, pip, cargo, go modules)
+    - Git operations (clone, commit, push, branch)
+    - Browser control (open, navigate, screenshot, interact)
+    - Process management (start, monitor, kill)
+    """
+
+    def __init__(self, workspace_root: str = "/app/litehat-workspace"):
+        self.workspace_root = Path(workspace_root)
+        self.workspace_root.mkdir(parents=True, exist_ok=True)
+
+        # Operation history for rollback
+        self.operation_history: List[FileOperation] = []
+
+        # Track active processes
+        self.active_processes: Dict[int, subprocess.Popen] = {}
+
+    # ── FILESYSTEM OPERATIONS ──
+
+    def read_file(self, path: str) -> str:
+        """Read a file from the workspace."""
+        full_path = self._resolve_path(path)
+        return full_path.read_text()
+
+    def write_file(self, path: str, content: str):
+        """Write a file, creating directories as needed. Backs up for rollback."""
+        full_path = self._resolve_path(path)
+        full_path.parent.mkdir(parents=True, exist_ok=True)
+
+        # Backup for rollback
+        if full_path.exists():
+            self.operation_history.append(FileOperation(
+                path=str(full_path),
+                operation="modify",
+                backup_content=full_path.read_text(),
+                timestamp=__import__('time').time(),
+            ))
+
+        full_path.write_text(content)
+
+    def delete_file(self, path: str):
+        """Delete a file, backing up for rollback."""
+        full_path = self._resolve_path(path)
+        if full_path.exists():
+            self.operation_history.append(FileOperation(
+                path=str(full_path),
+                operation="delete",
+                backup_content=full_path.read_text(),
+                timestamp=__import__('time').time(),
+            ))
+            full_path.unlink()
+
+    def list_directory(self, path: str = ".") -> List[Dict[str, Any]]:
+        """List directory contents."""
+        full_path = self._resolve_path(path)
+        entries = []
+        for entry in full_path.iterdir():
+            entries.append({
+                "name": entry.name,
+                "type": "directory" if entry.is_dir() else "file",
+                "size": entry.stat().st_size if entry.is_file() else 0,
+            })
+        return entries
+
+    def traverse_project(self) -> Dict[str, Any]:
+        """Get full project structure for the AI to understand the codebase."""
+        structure = {}
+        for root, dirs, files in os.walk(self.workspace_root):
+            # Skip hidden and node_modules
+            dirs[:] = [d for d in dirs if not d.startswith('.') and d != 'node_modules']
+
+            rel_path = os.path.relpath(root, self.workspace_root)
+            if rel_path == '.':
+                rel_path = ''
+
+            structure[rel_path] = {
+                "dirs": dirs,
+                "files": files,
+            }
+        return structure
+
+    # ── SHELL EXECUTION ──
+
+    def execute(
+        self,
+        command: str,
+        cwd: Optional[str] = None,
+        timeout: int = 300,
+        env: Optional[Dict[str, str]] = None,
+    ) -> TerminalResult:
+        """Execute a shell command and return the result."""
+        import time
+        start = time.time()
+
+        work_dir = self._resolve_path(cwd or ".").resolve()
+
+        merged_env = os.environ.copy()
+        if env:
+            merged_env.update(env)
+
+        try:
+            result = subprocess.run(
+                command,
+                shell=True,
+                cwd=str(work_dir),
+                capture_output=True,
+                text=True,
+                timeout=timeout,
+                env=merged_env,
+            )
+            elapsed = (time.time() - start) * 1000
+
+            return TerminalResult(
+                command=command,
+                stdout=result.stdout,
+                stderr=result.stderr,
+                exit_code=result.returncode,
+                elapsed_ms=elapsed,
+                working_dir=str(work_dir),
+            )
+        except subprocess.TimeoutExpired:
+            elapsed = (time.time() - start) * 1000
+            return TerminalResult(
+                command=command,
+                stdout="",
+                stderr=f"Command timed out after {timeout}s",
+                exit_code=-1,
+                elapsed_ms=elapsed,
+                working_dir=str(work_dir),
+            )
+
+    def execute_streaming(self, command: str, cwd: Optional[str] = None):
+        """Execute a command with streaming output (generator)."""
+        work_dir = self._resolve_path(cwd or ".").resolve()
+
+        process = subprocess.Popen(
+            command,
+            shell=True,
+            cwd=str(work_dir),
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            text=True,
+            bufsize=1,
+        )
+
+        self.active_processes[id(process)] = process
+
+        for line in process.stdout:
+            yield line.strip()
+
+        process.wait()
+        del self.active_processes[id(process)]
+
+    # ── PACKAGE MANAGEMENT ──
+
+    def npm_install(self, packages: List[str], dev: bool = False):
+        """Install npm packages."""
+        pkg_str = " ".join(packages)
+        flag = "--save-dev" if dev else "--save"
+        return self.execute(f"npm install {flag} {pkg_str}")
+
+    def pip_install(self, packages: List[str]):
+        """Install pip packages."""
+        pkg_str = " ".join(packages)
+        return self.execute(f"pip install {pkg_str}")
+
+    def cargo_add(self, packages: List[str]):
+        """Add Cargo dependencies."""
+        pkg_str = " ".join(packages)
+        return self.execute(f"cargo add {pkg_str}")
+
+    # ── GIT OPERATIONS ──
+
+    def git_init(self):
+        """Initialize a git repository."""
+        return self.execute("git init")
+
+    def git_clone(self, url: str, branch: Optional[str] = None):
+        """Clone a git repository."""
+        branch_flag = f"-b {branch}" if branch else ""
+        return self.execute(f"git clone {branch_flag} {url}")
+
+    def git_commit(self, message: str):
+        """Stage all and commit."""
+        self.execute("git add -A")
+        return self.execute(f'git commit -m "{message}"')
+
+    def git_push(self, remote: str = "origin", branch: str = "main"):
+        """Push to remote."""
+        return self.execute(f"git push {remote} {branch}")
+
+    def git_create_branch(self, branch: str):
+        """Create and switch to a new branch."""
+        return self.execute(f"git checkout -b {branch}")
+
+    # ── BROWSER CONTROL ──
+
+    async def browser_open(self, url: str):
+        """Open a URL in the browser and return page content."""
+        # Uses Playwright for full browser control
+        try:
+            from playwright.async_api import async_playwright
+            async with async_playwright() as p:
+                browser = await p.chromium.launch()
+                page = await browser.new_page()
+                await page.goto(url)
+                content = await page.content()
+                await browser.close()
+                return content
+        except ImportError:
+            return self.execute(f"curl -sL {url}").stdout
+
+    async def browser_screenshot(self, url: str, output_path: str):
+        """Take a screenshot of a URL."""
+        try:
+            from playwright.async_api import async_playwright
+            async with async_playwright() as p:
+                browser = await p.chromium.launch()
+                page = await browser.new_page()
+                await page.goto(url)
+                await page.screenshot(path=output_path, full_page=True)
+                await browser.close()
+        except ImportError:
+            pass
+
+    # ── ROLLBACK ──
+
+    def rollback(self, num_operations: int = 1):
+        """Rollback the last N file operations."""
+        for _ in range(num_operations):
+            if not self.operation_history:
+                break
+            op = self.operation_history.pop()
+            path = Path(op.path)
+
+            if op.operation == "modify" and op.backup_content is not None:
+                path.write_text(op.backup_content)
+            elif op.operation == "delete" and op.backup_content is not None:
+                path.write_text(op.backup_content)
+            elif op.operation == "create":
+                if path.exists():
+                    path.unlink()
+
+    def full_rollback(self):
+        """Rollback ALL operations since last checkpoint."""
+        self.rollback(len(self.operation_history))
+
+    # ── HELPERS ──
+
+    def _resolve_path(self, path: str) -> Path:
+        """Resolve a relative path to the workspace root."""
+        p = Path(path)
+        if p.is_absolute():
+            return p
+        return self.workspace_root / p
+
+    def create_checkpoint(self, name: str):
+        """Create a named checkpoint for later rollback."""
+        checkpoint_dir = self.workspace_root / ".litehat" / "checkpoints" / name
+        checkpoint_dir.mkdir(parents=True, exist_ok=True)
+
+        # Save operation history
+        history_path = checkpoint_dir / "operations.json"
+        history_path.write_text(json.dumps([
+            {"path": op.path, "operation": op.operation, "backup": op.backup_content}
+            for op in self.operation_history
+        ]))
+
+    def restore_checkpoint(self, name: str):
+        """Restore from a named checkpoint."""
+        checkpoint_dir = self.workspace_root / ".litehat" / "checkpoints" / name
+        if not checkpoint_dir.exists():
+            return False
+
+        self.full_rollback()
+        return True