code/tools/fs.py

"""File system tools: read, write, edit, list, glob, grep.

Enhanced with Hugging Face Hub cloud storage integration:
- All file operations work seamlessly with both local and cloud storage
- Automatic sync to HF Hub when configured
- Transparent fallback between local and remote

All tools are sandboxed to a configurable workspace root (default: ./workspace).
They return JSON-serializable dicts so they can be exposed via the API.
"""

from __future__ import annotations

import fnmatch
import logging
import os
import re
import shutil
from pathlib import Path
from typing import Any, Optional

logger = logging.getLogger(__name__)

# ─── Workspace sandbox ──────────────────────────────────────────────────

# Default workspace: ./workspace under the app root
_DEFAULT_WORKSPACE = os.environ.get(
    "SONICODER_WORKSPACE",
    os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "workspace")),
)

# Hub storage instance (lazy initialized)
_hub_storage = None


def get_workspace_root() -> str:
    """Return the absolute path of the agent's workspace root."""
    root = _DEFAULT_WORKSPACE
    os.makedirs(root, exist_ok=True)
    return root


def _resolve_safe(path: str) -> str:
    """Resolve a path safely within the workspace root.

    Raises ValueError if the resolved path escapes the workspace.
    """
    root = get_workspace_root()
    if os.path.isabs(path):
        full = os.path.abspath(path)
    else:
        full = os.path.abspath(os.path.join(root, path))

    # Ensure path is within the workspace
    if not (full == root or full.startswith(root + os.sep)):
        raise ValueError(
            f"Path '{path}' resolves outside the workspace root ({root}). "
            "Agent tools are sandboxed."
        )
    return full


def _get_hub_storage():
    """Get the HubStorage instance if available."""
    global _hub_storage
    if _hub_storage is None:
        try:
            from code.hub_storage import get_hub_storage, is_hub_available
            if is_hub_available():
                _hub_storage = get_hub_storage()
                logger.debug("Hub storage integration active")
            else:
                logger.debug("Hub storage not configured")
        except ImportError:
            logger.debug("huggingface_hub not installed")
            _hub_storage = None
    return _hub_storage


# ─── read_file ──────────────────────────────────────────────────────────

def read_file(path: str, offset: int = 0, limit: int | None = None) -> dict[str, Any]:
    """Read a text file from the workspace.

    Enhanced: Falls back to Hub storage if file not found locally.

    Args:
        path: Relative path inside the workspace, or absolute within it.
        offset: 1-indexed line to start reading from.
        limit: Maximum number of lines to read.

    Returns:
        dict with: path, content, line_count, truncated
    """
    try:
        full = _resolve_safe(path)
        
        # Try local first
        if os.path.exists(full) and os.path.isfile(full):
            with open(full, "r", encoding="utf-8", errors="replace") as f:
                lines = f.readlines()
            
            total = len(lines)
            start = max(0, (offset - 1) if offset > 0 else 0)
            end = (start + limit) if limit else total
            selected = lines[start:end]

            # Re-number for display
            numbered = "".join(
                f"{start + i + 1:6}\t{line}" for i, line in enumerate(selected)
            )

            return {
                "success": True,
                "path": path,
                "content": numbered,
                "line_count": total,
                "returned_lines": len(selected),
                "truncated": end < total,
                "source": "local",
            }
        
        # Try Hub storage
        hub = _get_hub_storage()
        if hub and hub.file_exists(path):
            try:
                content = hub.read_file(path)
                lines = content.split('\n')
                
                total = len(lines)
                start = max(0, (offset - 1) if offset > 0 else 0)
                end = (start + limit) if limit else total
                selected = lines[start:end]
                
                numbered = "".join(
                    f"{start + i + 1:6}\t{line}" for i, line in enumerate(selected)
                )
                
                return {
                    "success": True,
                    "path": path,
                    "content": numbered,
                    "line_count": total,
                    "returned_lines": len(selected),
                    "truncated": end < total,
                    "source": "hub",
                }
            except Exception as e:
                logger.warning("Failed to read %s from Hub: %s", path, e)
        
        return {"success": False, "error": f"File not found: {path}"}
    
    except Exception as exc:
        return {"success": False, "error": str(exc)}


# ─── write_file ─────────────────────────────────────────────────────────

def write_file(path: str, content: str) -> dict[str, Any]:
    """Write content to a file, creating parent directories as needed.

    Enhanced: Automatically syncs to HF Hub when configured.
    
    Args:
        path: Relative file path in workspace
        content: File content string
        
    Returns:
        dict with success status and metadata
    """
    try:
        full = _resolve_safe(path)
        os.makedirs(os.path.dirname(full), exist_ok=True)
        
        # Write locally first
        with open(full, "w", encoding="utf-8") as f:
            f.write(content)
        
        result = {
            "success": True,
            "path": path,
            "bytes_written": len(content.encode("utf-8")),
            "source": "local",
        }
        
        # Sync to Hub if configured
        hub = _get_hub_storage()
        if hub:
            try:
                metadata = hub.write_file(
                    path=path,
                    content=content,
                    commit_message=f"AI generated/updated: {path}"
                )
                result["hub_sync"] = True
                result["hub_remote"] = metadata.is_remote
                result["source"] = "synced"
                logger.info("Synced %s to Hub", path)
            except Exception as e:
                logger.warning("Failed to sync %s to Hub: %s", path, e)
                result["hub_sync"] = False
                result["hub_error"] = str(e)
        
        return result
    
    except Exception as exc:
        return {"success": False, "error": str(exc)}


# ─── edit_file ──────────────────────────────────────────────────────────

def edit_file(
    path: str,
    old_str: str,
    new_str: str,
    replace_all: bool = False,
) -> dict[str, Any]:
    """Replace occurrences of old_str with new_str in a file.

    Enhanced: Syncs changes back to Hub after edit.
    """
    try:
        full = _resolve_safe(path)
        
        # Read file (local or Hub)
        hub = _get_hub_storage()
        if os.path.exists(full):
            with open(full, "r", encoding="utf-8") as f:
                content = f.read()
        elif hub and hub.file_exists(path):
            content = hub.read_file(path)
        else:
            return {"success": False, "error": f"File not found: {path}"}

        if old_str not in content:
            return {
                "success": False,
                "error": f"old_str not found in {path}. Edit aborted.",
            }

        if old_str == new_str:
            return {"success": False, "error": "old_str and new_str are identical."}

        count = content.count(old_str) if replace_all else 1
        if not replace_all and count > 1:
            return {
                "success": False,
                "error": (
                    f"old_str is not unique ({count} matches) in {path}. "
                    "Provide more context or use replace_all=true."
                ),
            }

        new_content = content.replace(old_str, new_str) if replace_all else content.replace(
            old_str, new_str, 1
        )

        # Write locally
        os.makedirs(os.path.dirname(full), exist_ok=True)
        with open(full, "w", encoding="utf-8") as f:
            f.write(new_content)

        result = {
            "success": True,
            "path": path,
            "replacements": count,
        }
        
        # Sync to Hub
        if hub:
            try:
                hub.write_file(path, new_content, commit_message=f"Edit: {path}")
                result["hub_sync"] = True
            except Exception as e:
                logger.warning("Failed to sync edit to Hub: %s", e)

        return result
    
    except Exception as exc:
        return {"success": False, "error": str(exc)}


def multi_edit(path: str, edits: list[dict[str, Any]]) -> dict[str, Any]:
    """Apply multiple edits to a file atomically (all-or-nothing).

    Enhanced: Uses batch operation for Hub sync.
    """
    try:
        full = _resolve_safe(path)
        
        # Read file
        hub = _get_hub_storage()
        if os.path.exists(full):
            with open(full, "r", encoding="utf-8") as f:
                content = f.read()
        elif hub and hub.file_exists(path):
            content = hub.read_file(path)
        else:
            return {"success": False, "error": f"File not found: {path}"}

        applied = 0
        for edit in edits:
            old_str = edit.get("old_str", "")
            new_str = edit.get("new_str", "")
            replace_all = edit.get("replace_all", False)
            if old_str not in content:
                return {
                    "success": False,
                    "error": f"old_str not found in {path} for edit #{applied + 1}.",
                    "applied": applied,
                }
            if old_str == new_str:
                return {
                    "success": False,
                    "error": f"old_str and new_str identical in edit #{applied + 1}.",
                    "applied": applied,
                }
            content = content.replace(old_str, new_str) if replace_all else content.replace(
                old_str, new_str, 1
            )
            applied += 1

        # Write locally
        os.makedirs(os.path.dirname(full), exist_ok=True)
        with open(full, "w", encoding="utf-8") as f:
            f.write(content)

        result = {"success": True, "path": path, "applied": applied}
        
        # Sync to Hub
        if hub:
            try:
                hub.write_file(path, content, commit_message=f"Multi-edit: {path}")
                result["hub_sync"] = True
            except Exception as e:
                logger.warning("Failed to sync multi-edit to Hub: %s", e)

        return result
    
    except Exception as exc:
        return {"success": False, "error": str(exc)}


# ─── list_dir ───────────────────────────────────────────────────────────

def list_dir(path: str = ".") -> dict[str, Any]:
    """List directory contents.

    Enhanced: Merges local and Hub files.
    """
    try:
        full = _resolve_safe(path)
        
        entries = []
        seen_names = set()
        
        # List local files
        if os.path.exists(full) and os.path.isdir(full):
            for name in sorted(os.listdir(full)):
                entry_path = os.path.join(full, name)
                stat = os.stat(entry_path)
                rel_path = os.path.relpath(entry_path, get_workspace_root())
                entries.append({
                    "name": name,
                    "type": "dir" if os.path.isdir(entry_path) else "file",
                    "size": stat.st_size,
                    "path": rel_path.replace(os.sep, "/"),
                    "source": "local",
                })
                seen_names.add(name)
        
        # Merge with Hub files
        hub = _get_hub_storage()
        if hub:
            try:
                hub_files = hub.list_files(path=path, recursive=False)
                for hf in hub_files:
                    hf_name = Path(hf["path"]).name
                    if hf_name not in seen_names:
                        entries.append({
                            "name": hf_name,
                            "type": "file",
                            "size": hf.get("size", 0),
                            "path": hf["path"],
                            "source": "hub",
                        })
                        seen_names.add(hf_name)
            except Exception as e:
                logger.warning("Failed to list Hub files: %s", e)

        return {
            "success": True,
            "path": path,
            "entries": sorted(entries, key=lambda x: x["name"]),
        }
    
    except Exception as exc:
        return {"success": False, "error": str(exc)}


# ─── glob ───────────────────────────────────────────────────────────────

def glob_paths(pattern: str, path: str = ".") -> dict[str, Any]:
    """Glob file paths matching a pattern, recursively.

    Enhanced: Searches both local and Hub storage.
    """
    try:
        full = _resolve_safe(path)
        matches: list[dict[str, Any]] = []
        seen_paths = set()
        
        # Local glob
        if os.path.exists(full):
            for root_dir, _dirs, files in os.walk(full):
                for fname in files:
                    if fnmatch.fnmatch(fname, pattern) or fnmatch.fnmatch(
                        os.path.relpath(os.path.join(root_dir, fname), full), pattern
                    ):
                        rel_path = os.path.relpath(os.path.join(root_dir, fname), get_workspace_root()).replace(os.sep, "/")
                        if rel_path not in seen_paths:
                            matches.append({"path": rel_path, "source": "local"})
                            seen_paths.add(rel_path)
        
        # Hub glob
        hub = _get_hub_storage()
        if hub:
            try:
                hub_files = hub.list_files(path=path, recursive=True)
                for hf in hub_files:
                    if fnmatch.fnmatch(Path(hf["path"]).name, pattern) or fnmatch.fnmatch(hf["path"], pattern):
                        if hf["path"] not in seen_paths:
                            matches.append({"path": hf["path"], "source": "hub"})
                            seen_paths.add(hf["path"])
            except Exception as e:
                logger.warning("Failed to glob Hub files: %s", e)
        
        matches.sort(key=lambda x: x["path"])
        return {"success": True, "pattern": pattern, "matches": matches}
    
    except Exception as exc:
        return {"success": False, "error": str(exc)}


# ─── grep ───────────────────────────────────────────────────────────────

def grep_search(
    pattern: str,
    path: str = ".",
    include: str | None = None,
    ignore_case: bool = False,
    max_results: int = 100,
) -> dict[str, Any]:
    """Search file contents with a regex pattern.

    Enhanced: Searches both local and Hub files.
    """
    try:
        flags = re.IGNORECASE if ignore_case else 0
        regex = re.compile(pattern, flags)

        matches: list[dict[str, Any]] = []
        
        # Search local files
        full = _resolve_safe(path)
        if os.path.exists(full):
            for root_dir, _dirs, files in os.walk(full):
                for fname in files:
                    if include and not fnmatch.fnmatch(fname, include):
                        continue
                    fpath = os.path.join(root_dir, fname)
                    try:
                        with open(fpath, "r", encoding="utf-8", errors="replace") as f:
                            for lineno, line in enumerate(f, 1):
                                if regex.search(line):
                                    matches.append({
                                        "file": os.path.relpath(fpath, get_workspace_root()).replace(os.sep, "/"),
                                        "line": lineno,
                                        "text": line.rstrip()[:500],
                                        "source": "local",
                                    })
                                    if len(matches) >= max_results:
                                        return {
                                            "success": True,
                                            "pattern": pattern,
                                            "matches": matches,
                                            "truncated": True,
                                        }
                    except (UnicodeDecodeError, PermissionError):
                        continue
        
        # Search Hub files
        hub = _get_hub_storage()
        if hub and len(matches) < max_results:
            try:
                hub_files = hub.list_files(path=path, recursive=True)
                for hf in hub_files:
                    if include and not fnmatch.fnmatch(Path(hf["path"]).name, include):
                        continue
                    try:
                        content = hub.read_file(hf["path"])
                        for lineno, line in enumerate(content.split('\n'), 1):
                            if regex.search(line):
                                matches.append({
                                    "file": hf["path"],
                                    "line": lineno,
                                    "text": line.rstrip()[:500],
                                    "source": "hub",
                                })
                                if len(matches) >= max_results:
                                    return {
                                        "success": True,
                                        "pattern": pattern,
                                        "matches": matches,
                                        "truncated": True,
                                    }
                    except Exception:
                        continue
            except Exception as e:
                logger.warning("Failed to search Hub files: %s", e)

        return {
            "success": True,
            "pattern": pattern,
            "matches": matches,
            "truncated": False,
        }
    
    except re.error as exc:
        return {"success": False, "error": f"Invalid regex: {exc}"}
    except Exception as exc:
        return {"success": False, "error": str(exc)}


# ─── Workspace management ───────────────────────────────────────────────

def list_workspace_tree(max_depth: int = 3) -> dict[str, Any]:
    """Return a tree view of the workspace.

    Enhanced: Includes Hub files.
    """
    try:
        root = get_workspace_root()

        def _walk(path: str, depth: int) -> dict[str, Any]:
            if depth > max_depth:
                return {"name": os.path.basename(path), "type": "dir", "truncated": True}
            entries = []
            try:
                for name in sorted(os.listdir(path)):
                    full = os.path.join(path, name)
                    if os.path.isdir(full):
                        entries.append(_walk(full, depth + 1))
                    else:
                        entries.append({
                            "name": name,
                            "type": "file",
                            "size": os.path.getsize(full),
                        })
            except PermissionError:
                pass
            return {"name": os.path.basename(path), "type": "dir", "children": entries}

        tree = _walk(root, 0)
        
        # Add Hub info
        hub = _get_hub_storage()
        if hub:
            stats = hub.get_storage_stats()
            tree["hub_enabled"] = stats["hub_enabled"]
            tree["storage_mode"] = stats["mode"]
        
        return {"success": True, "tree": tree}
    
    except Exception as exc:
        return {"success": False, "error": str(exc)}


def reset_workspace() -> dict[str, Any]:
    """Clear all files in the workspace (used by /new command).

    Note: Does NOT clear Hub storage - use reset_hub_workspace for that.
    """
    try:
        root = get_workspace_root()
        if os.path.exists(root):
            for entry in os.listdir(root):
                full = os.path.join(root, entry)
                if os.path.isdir(full):
                    shutil.rmtree(full)
                else:
                    os.remove(full)
        return {"success": True, "message": "Workspace cleared"}
    except Exception as exc:
        return {"success": False, "error": str(exc)}


def snapshot_workspace() -> dict[str, str]:
    """Return a dict of {relative_path: content} for all text files in the workspace.

    Enhanced: Includes Hub files in snapshot.
    Used to package workspace files for ZIP/HF deploy.
    """
    root = get_workspace_root()
    files: dict[str, str] = {}
    
    # Local files
    for dirpath, _dirs, fnames in os.walk(root):
        parts = os.path.relpath(dirpath, root).split(os.sep)
        if any(p.startswith(".") or p in {"node_modules", "__pycache__", ".venv", "venv"} for p in parts):
            continue
        for fname in fnames:
            if fname.startswith("."):
                continue
            full = os.path.join(dirpath, fname)
            try:
                with open(full, "r", encoding="utf-8") as f:
                    files[os.path.relpath(full, root)] = f.read()
            except (UnicodeDecodeError, PermissionError):
                continue
    
    # Hub files (merge, don't overwrite local)
    hub = _get_hub_storage()
    if hub:
        try:
            hub_files = hub.list_files()
            for hf in hub_files:
                if hf["path"] not in files:
                    try:
                        content = hub.read_file(hf["path"])
                        files[hf["path"]] = content
                    except Exception:
                        continue
        except Exception as e:
            logger.warning("Failed to snapshot Hub files: %s", e)
    
    return files


# ─── Hub-specific operations ────────────────────────────────────────────

def sync_to_hub(path: str | None = None) -> dict[str, Any]:
    """Manually trigger sync to HF Hub.
    
    Args:
        path: Specific file/directory to sync (None = all)
        
    Returns:
        Sync result dict
    """
    hub = _get_hub_storage()
    if not hub:
        return {"success": False, "error": "Hub storage not configured"}
    
    try:
        return hub.sync_to_hub(path=path)
    except Exception as e:
        return {"success": False, "error": str(e)}


def sync_from_hub(path: str | None = None) -> dict[str, Any]:
    """Manually trigger sync from HF Hub.
    
    Args:
        path: Specific file to download (None = all)
        
    Returns:
        Sync result dict
    """
    hub = _get_hub_storage()
    if not hub:
        return {"success": False, "error": "Hub storage not configured"}
    
    try:
        return hub.sync_from_hub(path=path)
    except Exception as e:
        return {"success": False, "error": str(e)}


def get_hub_status() -> dict[str, Any]:
    """Get current Hub storage status and configuration."""
    hub = _get_hub_storage()
    if not hub:
        return {
            "available": False,
            "configured": False,
            "mode": "local_only",
        }
    
    try:
        stats = hub.get_storage_stats()
        return {
            "available": True,
            "configured": True,
            **stats,
        }
    except Exception as e:
        return {
            "available": False,
            "configured": True,
            "error": str(e),
        }


def delete_from_hub(path: str) -> dict[str, Any]:
    """Delete a file from Hub storage only (keeps local).
    
    Args:
        path: File path to delete from Hub
        
    Returns:
        Operation result
    """
    hub = _get_hub_storage()
    if not hub:
        return {"success": False, "error": "Hub storage not configured"}
    
    try:
        success = hub.delete_file(path, commit_message=f"Delete: {path}")
        return {"success": success, "path": path}
    except Exception as e:
        return {"success": False, "error": str(e)}