core/engine/src/tools/file_ops.py - main - alexefan136/flowstack

core/engine/src/tools/file_ops.py
1 922 строки70 KB
Alexander Efanov
Обновление репозитория
15 июл 2026, 12:19
15 июл 2026, 12:1976704c6
"""
File Operations Tool — MCP tool для безопасной работы с файловой системой.

Реализует MCP Tool specification 2024-11-05:
https://modelcontextprotocol.io/specification/2024-11-05/server/tools

Поддерживаемые операции:

**Файлы:**
- `read` — чтение файла (текст или base64 для бинарных)
- `write` — запись/перезапись файла
- `append` — добавление в конец файла
- `delete` — удаление файла
- `move` — перемещение/переименование
- `copy` — копирование
- `stat` — информация о файле (размер, права, даты)
- `exists` — проверка существования

**Директории:**
- `list` — листинг содержимого
- `mkdir` — создание директории (рекурсивно)
- `rmdir` — удаление пустой директории
- `tree` — дерево директории (с ограничениями по глубине)

**Поиск:**
- `search` — поиск файлов по glob паттерну

Архитектура безопасности (defense in depth):
1. **Sandbox (base_dir)** — все операции только внутри разрешённой директории
2. **Path traversal protection** — resolve() + проверка containment
3. **Symlink escape protection** — запрет ссылок за пределы sandbox
4. **File size limits** — защита от OOM при чтении больших файлов
5. **Blacklist опасных путей** — .git, node_modules, __pycache__ и т.д.
6. **Read-only mode** — опциональный запрет write операций
7. **Binary file detection** — корректная обработка бинарных файлов через base64

Примеры:
    file_ops(operation="read", path="notes/todo.md")
    file_ops(operation="list", path="projects/", recursive=False)
    file_ops(operation="write", path="output/result.json", content="{...}")
    file_ops(operation="search", path=".", pattern="**/*.py", recursive=True)
"""
from __future__ import annotations

import asyncio
import base64
import logging
import mimetypes
import os
import shutil
import stat
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any, ClassVar

from src.tools.base import Content, ImageContent, TextContent, Tool, ToolResult


logger = logging.getLogger(__name__)


# ============================================================================
# Exceptions
# ============================================================================


class FileOpsError(Exception):
    """Базовое исключение для file operations."""


class PathTraversalError(FileOpsError):
    """Попытка выйти за пределы sandbox."""


class FileTooLargeError(FileOpsError):
    """Файл превышает лимит размера."""


class ForbiddenPathError(FileOpsError):
    """Доступ к запрещённому пути."""


class ReadOnlyError(FileOpsError):
    """Попытка write операции в read-only режиме."""


class SymlinkEscapeError(FileOpsError):
    """Symlink выходит за пределы sandbox."""


class OperationNotPermittedError(FileOpsError):
    """Операция не разрешена конфигурацией."""


# ============================================================================
# Configuration
# ============================================================================


@dataclass
class FileOpsConfig:
    """
    Конфигурация для file operations tool.

    Все пути разрешаются относительно base_dir (sandbox).
    """

    # Корневая директория (sandbox) — обязательна
    base_dir: str = "."

    # Безопасность
    read_only: bool = False
    allow_symlinks: bool = False  # Разрешить symlinks внутри sandbox
    follow_symlinks: bool = True  # Разрешать symlinks при чтении

    # Лимиты
    max_file_size_mb: float = 10.0  # Макс размер файла для чтения
    max_write_size_mb: float = 50.0  # Макс размер записи
    max_listing_items: int = 1000  # Макс элементов в list/tree
    max_tree_depth: int = 5  # Макс глубина для tree
    max_search_results: int = 500  # Макс результатов search

    # Чёрный список паттернов путей (относительно base_dir)
    blocked_paths: list[str] = field(
        default_factory=lambda: [
            ".git",
            ".svn",
            ".hg",
            "node_modules",
            "__pycache__",
            "*.pyc",
            ".DS_Store",
            "Thumbs.db",
            ".env",
            ".env.*",
            "*.key",
            "*.pem",
            "id_rsa",
            "id_ed25519",
            ".ssh",
            ".gnupg",
        ]
    )

    # Разрешённые расширения для чтения (None = все)
    allowed_read_extensions: list[str] | None = None

    # Запрещённые расширения для записи
    blocked_write_extensions: list[str] = field(
        default_factory=lambda: [".exe", ".dll", ".so", ".dylib", ".bin"]
    )

    # Кодировка для текстовых файлов
    default_encoding: str = "utf-8"

    # Размер порога для определения binary файла
    binary_detection_size: int = 8192  # 8 KB

    def validate(self) -> list[str]:
        """Валидировать конфигурацию. Чистая функция."""
        errors: list[str] = []

        if not self.base_dir:
            errors.append("base_dir is required")

        if self.max_file_size_mb <= 0:
            errors.append("max_file_size_mb must be positive")

        if self.max_write_size_mb <= 0:
            errors.append("max_write_size_mb must be positive")

        if self.max_listing_items <= 0:
            errors.append("max_listing_items must be positive")

        if self.max_tree_depth < 0:
            errors.append("max_tree_depth must be non-negative")

        return errors


# ============================================================================
# Response Models
# ============================================================================


@dataclass
class FileInfo:
    """Информация о файле/директории."""

    path: str  # Относительный путь от base_dir
    absolute_path: str
    name: str
    is_file: bool
    is_directory: bool
    is_symlink: bool
    size: int  # bytes
    size_human: str
    permissions: str  # например "rwxr-xr-x"
    owner_uid: int
    owner_gid: int
    modified_at: datetime | None
    accessed_at: datetime | None
    created_at: datetime | None
    mime_type: str | None
    extension: str

    def to_dict(self) -> dict[str, Any]:
        """Чистая функция."""
        return {
            "path": self.path,
            "absolute_path": self.absolute_path,
            "name": self.name,
            "is_file": self.is_file,
            "is_directory": self.is_directory,
            "is_symlink": self.is_symlink,
            "size": self.size,
            "size_human": self.size_human,
            "permissions": self.permissions,
            "owner_uid": self.owner_uid,
            "owner_gid": self.owner_gid,
            "modified_at": (
                self.modified_at.isoformat() if self.modified_at else None
            ),
            "accessed_at": (
                self.accessed_at.isoformat() if self.accessed_at else None
            ),
            "created_at": (
                self.created_at.isoformat() if self.created_at else None
            ),
            "mime_type": self.mime_type,
            "extension": self.extension,
        }


@dataclass
class DirectoryListing:
    """Результат листинга директории."""

    path: str
    files: list[FileInfo]
    directories: list[FileInfo]
    total_items: int
    truncated: bool

    def to_dict(self) -> dict[str, Any]:
        """Чистая функция."""
        return {
            "path": self.path,
            "files": [f.to_dict() for f in self.files],
            "directories": [d.to_dict() for d in self.directories],
            "files_count": len(self.files),
            "directories_count": len(self.directories),
            "total_items": self.total_items,
            "truncated": self.truncated,
        }


# ============================================================================
# Helper Functions
# ============================================================================


def _format_size(size_bytes: int) -> str:
    """Форматировать размер в человекочитаемый вид. Чистая функция."""
    if size_bytes < 0:
        return "0 B"
    if size_bytes == 0:
        return "0 B"

    units = ["B", "KB", "MB", "GB", "TB"]
    size = float(size_bytes)
    unit_index = 0
    while size >= 1024.0 and unit_index < len(units) - 1:
        size /= 1024.0
        unit_index += 1

    if unit_index == 0:
        return f"{int(size)} B"
    return f"{size:.2f} {units[unit_index]}"


def _format_permissions(mode: int) -> str:
    """Форматировать Unix permissions как строку. Чистая функция."""
    perms = ""
    for who in ("USR", "GRP", "OTH"):
        for what, letter in (("R", "r"), ("W", "w"), ("X", "x")):
            flag = getattr(stat, f"S_I{what}{who}")
            perms += letter if mode & flag else "-"
    return perms


def _is_binary_data(data: bytes) -> bool:
    """
    Определить, являются ли данные бинарными.

    Чистая функция — проверяет наличие null bytes.
    """
    # Проверка на null bytes — классический признак binary
    return b"\x00" in data


def _guess_mime_type(path: Path) -> str | None:
    """Определить MIME type по расширению. Чистая функция."""
    mime_type, _ = mimetypes.guess_type(str(path))
    return mime_type


def _timestamp_to_datetime(ts: float) -> datetime | None:
    """Конвертировать Unix timestamp в datetime. Чистая функция."""
    if ts <= 0:
        return None
    try:
        return datetime.fromtimestamp(ts)
    except (ValueError, OSError, OverflowError):
        return None


# ============================================================================
# Path Validator
# ============================================================================


class PathValidator:
    """
    Валидатор путей для защиты от traversal и symlink escape.

    Обеспечивает что все операции остаются внутри base_dir (sandbox).
    """

    def __init__(self, config: FileOpsConfig):
        self.config = config
        # Resolve base_dir в абсолютный путь
        self._base_dir = Path(config.base_dir).resolve()

    @property
    def base_dir(self) -> Path:
        """Получить абсолютный путь к base directory."""
        return self._base_dir

    def validate_and_resolve(self, user_path: str) -> Path:
        """
        Валидировать и разрешить пользовательский путь.

        Args:
            user_path: Путь, введённый пользователем (может быть относительным).

        Returns:
            Абсолютный Path, гарантированно внутри sandbox.

        Raises:
            PathTraversalError: если путь выходит за пределы sandbox.
            ForbiddenPathError: если путь в чёрном списке.
            SymlinkEscapeError: если symlink выходит за пределы sandbox.
        """
        if not user_path:
            # Пустой путь = base_dir
            return self._base_dir

        # Нормализация: убираем leading slash для объединения с base
        normalized = user_path.lstrip("/").lstrip("\\")

        # Собираем путь относительно base_dir
        # НЕ используем resolve() сразу — сначала проверяем на .. вручную
        candidate = self._base_dir / normalized

        # Проверяем наличие .. в исходном пути (до resolve)
        # Это защита от некоторых edge cases
        parts = Path(normalized).parts
        if ".." in parts:
            # Разрешаем resolve() и потом проверяем containment
            pass

        # Resolve убирает .. и symlinks
        if self.config.follow_symlinks:
            resolved = candidate.resolve()
        else:
            # Не follow symlinks — используем absolute()
            resolved = candidate.absolute()

        # Проверка containment — путь должен быть внутри base_dir
        try:
            resolved.relative_to(self._base_dir)
        except ValueError:
            raise PathTraversalError(
                f"Path escapes sandbox: {user_path!r}. "
                f"All paths must be within {self._base_dir}"
            )

        # Проверка на symlink escape (если allow_symlinks=False)
        if not self.config.allow_symlinks and candidate.is_symlink():
            link_target = candidate.resolve()
            try:
                link_target.relative_to(self._base_dir)
            except ValueError:
                raise SymlinkEscapeError(
                    f"Symlink {user_path!r} points outside sandbox: {link_target}"
                )

        # Проверка на blacklisted paths
        self._check_blocked_paths(resolved)

        return resolved

    def _check_blocked_paths(self, resolved: Path) -> None:
        """Проверить путь на совпадение с чёрным списком."""
        # Относительный путь от base_dir
        try:
            rel_path = resolved.relative_to(self._base_dir)
        except ValueError:
            return  # Уже проверено в validate_and_resolve

        rel_str = str(rel_path)
        parts = rel_path.parts

        import fnmatch

        for pattern in self.config.blocked_paths:
            # Проверяем каждую часть пути
            for part in parts:
                if fnmatch.fnmatch(part, pattern):
                    raise ForbiddenPathError(
                        f"Path {rel_str!r} matches blocked pattern {pattern!r}"
                    )

            # Проверяем весь путь
            if fnmatch.fnmatch(rel_str, pattern):
                raise ForbiddenPathError(
                    f"Path {rel_str!r} matches blocked pattern {pattern!r}"
                )

    def to_relative(self, absolute_path: Path) -> str:
        """
        Конвертировать абсолютный путь в относительный от base_dir.

        Чистая функция (предполагает что путь уже валидирован).
        """
        try:
            return str(absolute_path.relative_to(self._base_dir))
        except ValueError:
            return str(absolute_path)


# ============================================================================
# File Operations Client
# ============================================================================


class FileOpsClient:
    """
    Клиент для операций с файловой системой.

    Использует asyncio.to_thread для асинхронного выполнения
    blocking file I/O операций.
    """

    def __init__(self, config: FileOpsConfig):
        self.config = config
        self.validator = PathValidator(config)

    @property
    def base_dir(self) -> Path:
        """Получить base directory."""
        return self.validator.base_dir

    def _check_read_only(self, operation: str) -> None:
        """Проверить, разрешена ли операция в read-only режиме."""
        write_ops = {
            "write", "append", "delete", "move", "copy",
            "mkdir", "rmdir",
        }
        if self.config.read_only and operation in write_ops:
            raise ReadOnlyError(
                f"Operation '{operation}' is not allowed in read-only mode"
            )

    def _check_write_extension(self, path: Path) -> None:
        """Проверить расширение файла для write операций."""
        ext = path.suffix.lower()
        if ext in self.config.blocked_write_extensions:
            raise OperationNotPermittedError(
                f"Writing files with extension {ext!r} is not allowed. "
                f"Blocked: {self.config.blocked_write_extensions}"
            )

    def _check_read_extension(self, path: Path) -> None:
        """Проверить расширение файла для read операций (если whitelist задан)."""
        if self.config.allowed_read_extensions is None:
            return
        ext = path.suffix.lower()
        if ext and ext not in self.config.allowed_read_extensions:
            raise OperationNotPermittedError(
                f"Reading files with extension {ext!r} is not allowed. "
                f"Allowed: {self.config.allowed_read_extensions}"
            )

    async def _run_in_thread(self, func, *args, **kwargs):
        """Запустить blocking функцию в thread pool."""
        return await asyncio.to_thread(func, *args, **kwargs)

    def _get_file_info(self, absolute_path: Path) -> FileInfo:
        """
        Получить информацию о файле/директории.

        Blocking функция — должна вызываться через _run_in_thread.
        """
        st = absolute_path.stat(follow_symlinks=self.config.follow_symlinks)

        is_symlink = absolute_path.is_symlink()
        is_file = absolute_path.is_file()
        is_directory = absolute_path.is_dir()

        mime_type = _guess_mime_type(absolute_path) if is_file else None

        return FileInfo(
            path=self.validator.to_relative(absolute_path),
            absolute_path=str(absolute_path),
            name=absolute_path.name,
            is_file=is_file,
            is_directory=is_directory,
            is_symlink=is_symlink,
            size=st.st_size,
            size_human=_format_size(st.st_size),
            permissions=_format_permissions(st.st_mode),
            owner_uid=st.st_uid,
            owner_gid=st.st_gid,
            modified_at=_timestamp_to_datetime(st.st_mtime),
            accessed_at=_timestamp_to_datetime(st.st_atime),
            created_at=_timestamp_to_datetime(st.st_ctime),
            mime_type=mime_type,
            extension=absolute_path.suffix.lower(),
        )

    # ====================================================================
    # Read Operations
    # ====================================================================

    async def read_file(self, user_path: str) -> dict[str, Any]:
        """
        Прочитать содержимое файла.

        Для текстовых файлов возвращает content как строку.
        Для бинарных — base64-encoded строку.
        """
        self._check_read_only("read")
        absolute_path = self.validator.validate_and_resolve(user_path)

        if not absolute_path.exists():
            raise FileNotFoundError(f"File not found: {user_path}")

        if not absolute_path.is_file():
            raise FileOpsError(f"Not a file: {user_path}")

        self._check_read_extension(absolute_path)

        # Проверка размера
        size = await self._run_in_thread(lambda: absolute_path.stat().st_size)
        max_bytes = int(self.config.max_file_size_mb * 1024 * 1024)
        if size > max_bytes:
            raise FileTooLargeError(
                f"File too large: {_format_size(size)} "
                f"(max: {_format_size(max_bytes)})"
            )

        # Читаем как bytes
        data: bytes = await self._run_in_thread(
            lambda: absolute_path.read_bytes()
        )

        # Определяем тип
        is_binary = _is_binary_data(data)
        mime_type = _guess_mime_type(absolute_path) or "application/octet-stream"

        info = await self._run_in_thread(
            lambda: self._get_file_info(absolute_path)
        )

        if is_binary:
            # Бинарный файл — base64
            encoded = base64.b64encode(data).decode("ascii")
            return {
                "operation": "read",
                "path": info.path,
                "is_binary": True,
                "encoding": "base64",
                "mime_type": mime_type,
                "size": size,
                "size_human": _format_size(size),
                "content": encoded,
                "content_length": len(encoded),
            }
        else:
            # Текстовый файл — декодируем
            try:
                text = data.decode(self.config.default_encoding)
                encoding_used = self.config.default_encoding
            except UnicodeDecodeError:
                # Fallback на latin-1 (never fails)
                text = data.decode("latin-1")
                encoding_used = "latin-1"

            lines = text.count("\n") + (1 if text and not text.endswith("\n") else 0)

            return {
                "operation": "read",
                "path": info.path,
                "is_binary": False,
                "encoding": encoding_used,
                "mime_type": mime_type,
                "size": size,
                "size_human": _format_size(size),
                "content": text,
                "lines": lines,
                "characters": len(text),
            }

    async def file_exists(self, user_path: str) -> dict[str, Any]:
        """Проверить существование файла/директории."""
        self._check_read_only("exists")
        absolute_path = self.validator.validate_and_resolve(user_path)

        exists = await self._run_in_thread(lambda: absolute_path.exists())
        is_file = False
        is_dir = False
        is_symlink = False

        if exists:
            is_file = await self._run_in_thread(lambda: absolute_path.is_file())
            is_dir = await self._run_in_thread(lambda: absolute_path.is_dir())
            is_symlink = await self._run_in_thread(
                lambda: absolute_path.is_symlink()
            )

        return {
            "operation": "exists",
            "path": self.validator.to_relative(absolute_path),
            "exists": exists,
            "is_file": is_file,
            "is_directory": is_dir,
            "is_symlink": is_symlink,
        }

    async def file_stat(self, user_path: str) -> dict[str, Any]:
        """Получить информацию о файле/директории."""
        self._check_read_only("stat")
        absolute_path = self.validator.validate_and_resolve(user_path)

        if not await self._run_in_thread(lambda: absolute_path.exists()):
            raise FileNotFoundError(f"Path not found: {user_path}")

        info = await self._run_in_thread(
            lambda: self._get_file_info(absolute_path)
        )

        return {
            "operation": "stat",
            **info.to_dict(),
        }

    async def list_directory(
        self,
        user_path: str,
        recursive: bool = False,
        show_hidden: bool = False,
    ) -> dict[str, Any]:
        """Листинг содержимого директории."""
        self._check_read_only("list")
        absolute_path = self.validator.validate_and_resolve(user_path)

        if not await self._run_in_thread(lambda: absolute_path.exists()):
            raise FileNotFoundError(f"Directory not found: {user_path}")

        if not await self._run_in_thread(lambda: absolute_path.is_dir()):
            raise FileOpsError(f"Not a directory: {user_path}")

        files: list[FileInfo] = []
        directories: list[FileInfo] = []
        truncated = False
        total_count = 0

        def collect_items() -> None:
            nonlocal truncated, total_count

            if recursive:
                iterator = absolute_path.rglob("*")
            else:
                iterator = absolute_path.iterdir()

            for item in iterator:
                total_count += 1
                if total_count > self.config.max_listing_items:
                    truncated = True
                    break

                # Пропускаем hidden files если не нужно
                if not show_hidden and item.name.startswith("."):
                    continue

                # Пропускаем blacklisted
                try:
                    self.validator._check_blocked_paths(item)
                except ForbiddenPathError:
                    continue

                try:
                    info = self._get_file_info(item)
                    if info.is_directory:
                        directories.append(info)
                    else:
                        files.append(info)
                except (OSError, PermissionError):
                    # Пропускаем недоступные файлы
                    continue

        await self._run_in_thread(collect_items)

        # Сортировка
        files.sort(key=lambda f: f.name.lower())
        directories.sort(key=lambda d: d.name.lower())

        listing = DirectoryListing(
            path=self.validator.to_relative(absolute_path),
            files=files,
            directories=directories,
            total_items=total_count,
            truncated=truncated,
        )

        return {
            "operation": "list",
            **listing.to_dict(),
        }

    async def tree_directory(
        self,
        user_path: str,
        max_depth: int | None = None,
        show_hidden: bool = False,
    ) -> dict[str, Any]:
        """Построить дерево директории."""
        self._check_read_only("tree")
        absolute_path = self.validator.validate_and_resolve(user_path)

        if not await self._run_in_thread(lambda: absolute_path.is_dir()):
            raise FileOpsError(f"Not a directory: {user_path}")

        effective_depth = (
            max_depth if max_depth is not None else self.config.max_tree_depth
        )
        effective_depth = min(effective_depth, self.config.max_tree_depth)

        lines: list[str] = []
        items_count = 0
        truncated = False

        def build_tree(current: Path, prefix: str, depth: int) -> None:
            nonlocal items_count, truncated

            if depth > effective_depth or truncated:
                return

            try:
                entries = sorted(
                    [e for e in current.iterdir()],
                    key=lambda e: (not e.is_dir(), e.name.lower()),
                )
            except (OSError, PermissionError):
                return

            # Фильтруем
            filtered: list[Path] = []
            for entry in entries:
                if not show_hidden and entry.name.startswith("."):
                    continue
                try:
                    self.validator._check_blocked_paths(entry)
                    filtered.append(entry)
                except ForbiddenPathError:
                    continue

            for i, entry in enumerate(filtered):
                items_count += 1
                if items_count > self.config.max_listing_items:
                    truncated = True
                    lines.append(f"{prefix}... (truncated)")
                    return

                is_last = i == len(filtered) - 1
                connector = "└── " if is_last else "├── "
                suffix = "/" if entry.is_dir() else ""
                lines.append(f"{prefix}{connector}{entry.name}{suffix}")

                if entry.is_dir():
                    extension = "    " if is_last else "│   "
                    build_tree(entry, prefix + extension, depth + 1)

        root_name = absolute_path.name or str(absolute_path)
        lines.append(f"{root_name}/")
        await self._run_in_thread(lambda: build_tree(absolute_path, "", 1))

        return {
            "operation": "tree",
            "path": self.validator.to_relative(absolute_path),
            "tree": "\n".join(lines),
            "items_count": items_count,
            "max_depth": effective_depth,
            "truncated": truncated,
        }

    async def search_files(
        self,
        user_path: str,
        pattern: str,
        recursive: bool = True,
    ) -> dict[str, Any]:
        """Поиск файлов по glob паттерну."""
        self._check_read_only("search")
        absolute_path = self.validator.validate_and_resolve(user_path)

        if not await self._run_in_thread(lambda: absolute_path.is_dir()):
            raise FileOpsError(f"Not a directory: {user_path}")

        results: list[dict[str, Any]] = []
        truncated = False

        def do_search() -> None:
            nonlocal truncated

            if recursive:
                glob_pattern = f"**/{pattern}"
            else:
                glob_pattern = pattern

            matches = list(absolute_path.glob(glob_pattern))

            for i, match in enumerate(matches):
                if i >= self.config.max_search_results:
                    truncated = True
                    break

                try:
                    self.validator._check_blocked_paths(match)
                except ForbiddenPathError:
                    continue

                try:
                    info = self._get_file_info(match)
                    results.append(info.to_dict())
                except (OSError, PermissionError):
                    continue

        await self._run_in_thread(do_search)

        return {
            "operation": "search",
            "path": self.validator.to_relative(absolute_path),
            "pattern": pattern,
            "recursive": recursive,
            "matches": results,
            "matches_count": len(results),
            "truncated": truncated,
        }

    # ====================================================================
    # Write Operations
    # ====================================================================

    async def write_file(
        self,
        user_path: str,
        content: str,
        encoding: str | None = None,
        append: bool = False,
    ) -> dict[str, Any]:
        """Записать или добавить в файл."""
        operation = "append" if append else "write"
        self._check_read_only(operation)

        absolute_path = self.validator.validate_and_resolve(user_path)
        self._check_write_extension(absolute_path)

        # Создаём родительские директории если нужно
        def ensure_parent() -> None:
            absolute_path.parent.mkdir(parents=True, exist_ok=True)

        await self._run_in_thread(ensure_parent)

        # Проверка размера контента
        content_bytes = content.encode(encoding or self.config.default_encoding)
        max_bytes = int(self.config.max_write_size_mb * 1024 * 1024)
        if len(content_bytes) > max_bytes:
            raise FileTooLargeError(
                f"Content too large: {_format_size(len(content_bytes))} "
                f"(max: {_format_size(max_bytes)})"
            )

        def do_write() -> None:
            if append:
                with open(
                    absolute_path,
                    "ab" if append else "wb",
                ) as f:
                    f.write(content_bytes)
            else:
                absolute_path.write_bytes(content_bytes)

        await self._run_in_thread(do_write)

        info = await self._run_in_thread(
            lambda: self._get_file_info(absolute_path)
        )

        return {
            "operation": operation,
            "path": info.path,
            "bytes_written": len(content_bytes),
            "bytes_written_human": _format_size(len(content_bytes)),
            "encoding": encoding or self.config.default_encoding,
            "final_size": info.size,
            "final_size_human": info.size_human,
        }

    async def delete_file(self, user_path: str) -> dict[str, Any]:
        """Удалить файл."""
        self._check_read_only("delete")
        absolute_path = self.validator.validate_and_resolve(user_path)

        if not await self._run_in_thread(lambda: absolute_path.exists()):
            raise FileNotFoundError(f"Path not found: {user_path}")

        # Собираем info до удаления
        info = await self._run_in_thread(
            lambda: self._get_file_info(absolute_path)
        )

        is_dir = info.is_directory

        if is_dir:
            # Проверяем что директория пустая
            is_empty = await self._run_in_thread(
                lambda: not any(absolute_path.iterdir())
            )
            if not is_empty:
                raise FileOpsError(
                    f"Directory not empty: {user_path}. "
                    "Use rmdir only for empty directories."
                )
            await self._run_in_thread(lambda: absolute_path.rmdir())
        else:
            await self._run_in_thread(lambda: absolute_path.unlink())

        return {
            "operation": "delete",
            "path": info.path,
            "was_directory": is_dir,
            "deleted_size": info.size,
            "deleted_size_human": info.size_human,
        }

    async def move_file(
        self, source_path: str, destination_path: str
    ) -> dict[str, Any]:
        """Переместить/переименовать файл или директорию."""
        self._check_read_only("move")

        source_abs = self.validator.validate_and_resolve(source_path)
        dest_abs = self.validator.validate_and_resolve(destination_path)

        if not await self._run_in_thread(lambda: source_abs.exists()):
            raise FileNotFoundError(f"Source not found: {source_path}")

        if await self._run_in_thread(lambda: dest_abs.exists()):
            raise FileOpsError(
                f"Destination already exists: {destination_path}"
            )

        self._check_write_extension(dest_abs)

        # Создаём родительские директории для destination
        def ensure_dest_parent() -> None:
            dest_abs.parent.mkdir(parents=True, exist_ok=True)

        await self._run_in_thread(ensure_dest_parent)

        source_info = await self._run_in_thread(
            lambda: self._get_file_info(source_abs)
        )

        await self._run_in_thread(lambda: shutil.move(str(source_abs), str(dest_abs)))

        return {
            "operation": "move",
            "source": source_info.path,
            "destination": self.validator.to_relative(dest_abs),
            "was_directory": source_info.is_directory,
            "size": source_info.size,
            "size_human": source_info.size_human,
        }

    async def copy_file(
        self, source_path: str, destination_path: str
    ) -> dict[str, Any]:
        """Копировать файл или директорию."""
        self._check_read_only("copy")

        source_abs = self.validator.validate_and_resolve(source_path)
        dest_abs = self.validator.validate_and_resolve(destination_path)

        if not await self._run_in_thread(lambda: source_abs.exists()):
            raise FileNotFoundError(f"Source not found: {source_path}")

        if await self._run_in_thread(lambda: dest_abs.exists()):
            raise FileOpsError(
                f"Destination already exists: {destination_path}"
            )

        self._check_write_extension(dest_abs)

        # Создаём родительские директории
        def ensure_dest_parent() -> None:
            dest_abs.parent.mkdir(parents=True, exist_ok=True)

        await self._run_in_thread(ensure_dest_parent)

        source_info = await self._run_in_thread(
            lambda: self._get_file_info(source_abs)
        )

        def do_copy() -> None:
            if source_abs.is_dir():
                shutil.copytree(str(source_abs), str(dest_abs))
            else:
                shutil.copy2(str(source_abs), str(dest_abs))

        await self._run_in_thread(do_copy)

        return {
            "operation": "copy",
            "source": source_info.path,
            "destination": self.validator.to_relative(dest_abs),
            "was_directory": source_info.is_directory,
            "size": source_info.size,
            "size_human": source_info.size_human,
        }

    async def make_directory(self, user_path: str) -> dict[str, Any]:
        """Создать директорию (рекурсивно)."""
        self._check_read_only("mkdir")
        absolute_path = self.validator.validate_and_resolve(user_path)

        if await self._run_in_thread(lambda: absolute_path.exists()):
            if await self._run_in_thread(lambda: absolute_path.is_dir()):
                return {
                    "operation": "mkdir",
                    "path": self.validator.to_relative(absolute_path),
                    "created": False,
                    "already_exists": True,
                }
            else:
                raise FileOpsError(
                    f"Path exists but is not a directory: {user_path}"
                )

        await self._run_in_thread(
            lambda: absolute_path.mkdir(parents=True, exist_ok=False)
        )

        return {
            "operation": "mkdir",
            "path": self.validator.to_relative(absolute_path),
            "created": True,
            "already_exists": False,
        }

    async def remove_directory(self, user_path: str) -> dict[str, Any]:
        """Удалить пустую директорию."""
        self._check_read_only("rmdir")
        absolute_path = self.validator.validate_and_resolve(user_path)

        if not await self._run_in_thread(lambda: absolute_path.exists()):
            raise FileNotFoundError(f"Directory not found: {user_path}")

        if not await self._run_in_thread(lambda: absolute_path.is_dir()):
            raise FileOpsError(f"Not a directory: {user_path}")

        # Проверяем что пустая
        is_empty = await self._run_in_thread(
            lambda: not any(absolute_path.iterdir())
        )
        if not is_empty:
            raise FileOpsError(
                f"Directory not empty: {user_path}. "
                "Only empty directories can be removed with rmdir."
            )

        rel_path = self.validator.to_relative(absolute_path)
        await self._run_in_thread(lambda: absolute_path.rmdir())

        return {
            "operation": "rmdir",
            "path": rel_path,
            "removed": True,
        }


# ============================================================================
# File Operations Tool (MCP)
# ============================================================================


class FileOpsTool(Tool):
    """
    MCP Tool для безопасных операций с файловой системой.

    Все пути разрешаются относительно base_dir (sandbox).
    Tool автоматически защищает от:
    - Path traversal атак
    - Symlink escape
    - Read-only violations
    - Oversized files
    - Blacklisted paths

    Операции:
    - **read**: чтение файла (текст/base64)
    - **write**: запись файла
    - **append**: добавление в файл
    - **delete**: удаление файла
    - **move**: перемещение/переименование
    - **copy**: копирование
    - **list**: листинг директории
    - **tree**: дерево директории
    - **mkdir**: создание директории
    - **rmdir**: удаление пустой директории
    - **stat**: информация о файле
    - **exists**: проверка существования
    - **search**: поиск по glob паттерну
    """

    name: ClassVar[str] = "file_ops"
    description: ClassVar[str] = (
        "Безопасные операции с файлами и директориями. "
        "Все пути — относительные от рабочей директории (sandbox). "
        "Операции: read, write, append, delete, move, copy, list, tree, "
        "mkdir, rmdir, stat, exists, search. "
        "Пример: {\"operation\": \"read\", \"path\": \"notes/todo.md\"}"
    )

    input_schema: ClassVar[dict[str, Any] | None] = {
        "type": "object",
        "properties": {
            "operation": {
                "type": "string",
                "enum": [
                    "read", "write", "append", "delete", "move", "copy",
                    "list", "tree", "mkdir", "rmdir",
                    "stat", "exists", "search",
                ],
                "description": "Тип операции",
            },
            "path": {
                "type": "string",
                "description": "Путь к файлу или директории (относительный от sandbox)",
            },
            "content": {
                "type": "string",
                "description": "Содержимое для write/append",
            },
            "destination": {
                "type": "string",
                "description": "Путь назначения для move/copy",
            },
            "pattern": {
                "type": "string",
                "description": "Glob паттерн для search (например, '**/*.py')",
            },
            "encoding": {
                "type": "string",
                "description": "Кодировка для write (по умолчанию utf-8)",
            },
            "recursive": {
                "type": "boolean",
                "default": False,
                "description": "Рекурсивно для list/search",
            },
            "show_hidden": {
                "type": "boolean",
                "default": False,
                "description": "Показывать hidden files (начинающиеся с .)",
            },
            "max_depth": {
                "type": "integer",
                "description": "Максимальная глубина для tree (по умолчанию из конфига)",
            },
        },
        "required": ["operation"],
        "additionalProperties": False,
    }

    parameters_schema: ClassVar[dict[str, Any] | None] = input_schema

    examples: ClassVar[list[dict[str, Any]]] = [
        {
            "operation": "read",
            "path": "notes/todo.md",
        },
        {
            "operation": "list",
            "path": "projects",
            "recursive": False,
        },
        {
            "operation": "write",
            "path": "output/result.txt",
            "content": "Hello, World!",
        },
        {
            "operation": "search",
            "path": ".",
            "pattern": "**/*.py",
            "recursive": True,
        },
        {
            "operation": "tree",
            "path": "src",
            "max_depth": 3,
        },
    ]

    tags: ClassVar[list[str]] = ["filesystem", "files", "utility"]
    is_read_only: ClassVar[bool] = False
    requires_confirmation: ClassVar[bool] = False

    # Операции, требующие path
    PATH_REQUIRED_OPS: ClassVar[set[str]] = {
        "read", "write", "append", "delete", "list", "tree",
        "mkdir", "rmdir", "stat", "exists", "search",
    }

    # Write операции (требуют read_only=False)
    WRITE_OPERATIONS: ClassVar[set[str]] = {
        "write", "append", "delete", "move", "copy", "mkdir", "rmdir",
    }

    # Dangerous операции (требуют extra caution)
    DESTRUCTIVE_OPS: ClassVar[set[str]] = {"delete", "rmdir"}

    def __init__(self, config: FileOpsConfig | None = None):
        self.config = config or FileOpsConfig()
        self._client: FileOpsClient | None = None

    def _get_client(self) -> FileOpsClient:
        """Получить или создать client."""
        if self._client is None:
            self._client = FileOpsClient(self.config)
        return self._client

    async def execute(self, **kwargs: Any) -> ToolResult:
        """
        Выполнить file operation.

        Args:
            operation: Тип операции
            path: Путь (обязателен для большинства операций)
            content: Содержимое (для write/append)
            destination: Путь назначения (для move/copy)
            pattern: Glob паттерн (для search)
            encoding: Кодировка (для write)
            recursive: Рекурсивно (для list/search)
            show_hidden: Показывать hidden files
            max_depth: Максимальная глубина (для tree)

        Returns:
            ToolResult с результатом операции.
        """
        operation_raw = kwargs.get("operation", "")
        if not isinstance(operation_raw, str) or not operation_raw:
            return ToolResult.failure("Parameter 'operation' is required")

        operation = operation_raw.strip().lower()

        # Валидация операции
        valid_ops = set(self.input_schema["properties"]["operation"]["enum"])  # type: ignore[index]
        if operation not in valid_ops:
            return ToolResult.failure(
                f"Unknown operation: {operation}. Valid: {', '.join(sorted(valid_ops))}"
            )

        # Получаем параметры
        path = kwargs.get("path")
        content = kwargs.get("content")
        destination = kwargs.get("destination")
        pattern = kwargs.get("pattern")
        encoding = kwargs.get("encoding")
        recursive = bool(kwargs.get("recursive", False))
        show_hidden = bool(kwargs.get("show_hidden", False))
        max_depth = kwargs.get("max_depth")

        # Проверка required параметров для каждой операции
        if operation in self.PATH_REQUIRED_OPS:
            if not path or not isinstance(path, str):
                return ToolResult.failure(
                    f"Parameter 'path' is required for operation '{operation}'"
                )

        if operation in ("write", "append"):
            if content is None or not isinstance(content, str):
                return ToolResult.failure(
                    f"Parameter 'content' (string) is required for operation '{operation}'"
                )

        if operation in ("move", "copy"):
            if not path or not isinstance(path, str):
                return ToolResult.failure(
                    f"Parameter 'path' is required for operation '{operation}'"
                )
            if not destination or not isinstance(destination, str):
                return ToolResult.failure(
                    f"Parameter 'destination' is required for operation '{operation}'"
                )

        if operation == "search":
            if not pattern or not isinstance(pattern, str):
                return ToolResult.failure(
                    "Parameter 'pattern' is required for operation 'search'"
                )

        try:
            client = self._get_client()

            # Dispatch по операции
            if operation == "read":
                result_data = await client.read_file(str(path))
            elif operation == "write":
                result_data = await client.write_file(
                    str(path), str(content), encoding=encoding, append=False
                )
            elif operation == "append":
                result_data = await client.write_file(
                    str(path), str(content), encoding=encoding, append=True
                )
            elif operation == "delete":
                result_data = await client.delete_file(str(path))
            elif operation == "move":
                result_data = await client.move_file(str(path), str(destination))
            elif operation == "copy":
                result_data = await client.copy_file(str(path), str(destination))
            elif operation == "list":
                result_data = await client.list_directory(
                    str(path), recursive=recursive, show_hidden=show_hidden
                )
            elif operation == "tree":
                result_data = await client.tree_directory(
                    str(path),
                    max_depth=int(max_depth) if max_depth is not None else None,
                    show_hidden=show_hidden,
                )
            elif operation == "mkdir":
                result_data = await client.make_directory(str(path))
            elif operation == "rmdir":
                result_data = await client.remove_directory(str(path))
            elif operation == "stat":
                result_data = await client.file_stat(str(path))
            elif operation == "exists":
                result_data = await client.file_exists(str(path))
            elif operation == "search":
                result_data = await client.search_files(
                    str(path), str(pattern), recursive=recursive
                )
            else:
                return ToolResult.failure(f"Unhandled operation: {operation}")

            # Формируем текстовый вывод
            text = self._format_result_text(operation, result_data)

            return ToolResult.success_result(
                [TextContent(text=text)],
                metadata=result_data,
            )

        except FileNotFoundError as e:
            return ToolResult.failure(
                f"File not found: {e}",
                metadata={"operation": operation},
            )
        except PathTraversalError as e:
            return ToolResult.failure(
                f"Security error: {e}",
                metadata={"operation": operation, "error_type": "path_traversal"},
            )
        except SymlinkEscapeError as e:
            return ToolResult.failure(
                f"Security error: {e}",
                metadata={"operation": operation, "error_type": "symlink_escape"},
            )
        except ForbiddenPathError as e:
            return ToolResult.failure(
                f"Forbidden path: {e}",
                metadata={"operation": operation, "error_type": "forbidden"},
            )
        except ReadOnlyError as e:
            return ToolResult.failure(
                str(e),
                metadata={"operation": operation, "error_type": "read_only"},
            )
        except FileTooLargeError as e:
            return ToolResult.failure(
                str(e),
                metadata={"operation": operation, "error_type": "file_too_large"},
            )
        except OperationNotPermittedError as e:
            return ToolResult.failure(
                str(e),
                metadata={"operation": operation, "error_type": "not_permitted"},
            )
        except PermissionError as e:
            return ToolResult.failure(
                f"Permission denied: {e}",
                metadata={"operation": operation, "error_type": "permission"},
            )
        except FileOpsError as e:
            return ToolResult.failure(
                f"File operation error: {e}",
                metadata={"operation": operation},
            )
        except Exception as e:
            logger.exception(f"Unexpected error in file_ops tool: {e}")
            return ToolResult.failure(
                f"Unexpected error: {type(e).__name__}: {e}",
                metadata={"operation": operation},
            )

    def _format_result_text(
        self, operation: str, result: dict[str, Any]
    ) -> str:
        """
        Форматировать результат в читаемый текст.

        Чистая функция.
        """
        if operation == "read":
            return self._format_read_result(result)
        elif operation in ("write", "append"):
            return self._format_write_result(operation, result)
        elif operation == "delete":
            return self._format_delete_result(result)
        elif operation in ("move", "copy"):
            return self._format_move_copy_result(operation, result)
        elif operation == "list":
            return self._format_list_result(result)
        elif operation == "tree":
            return self._format_tree_result(result)
        elif operation in ("mkdir", "rmdir"):
            return self._format_dir_op_result(operation, result)
        elif operation == "stat":
            return self._format_stat_result(result)
        elif operation == "exists":
            return self._format_exists_result(result)
        elif operation == "search":
            return self._format_search_result(result)
        else:
            import json
            return json.dumps(result, ensure_ascii=False, indent=2, default=str)

    def _format_read_result(self, result: dict[str, Any]) -> str:
        """Форматировать результат read."""
        path = result.get("path", "")
        is_binary = result.get("is_binary", False)
        size_human = result.get("size_human", "0 B")
        mime_type = result.get("mime_type", "unknown")

        lines: list[str] = [
            f"# File: {path}",
            "",
            f"- **Size:** {size_human}",
            f"- **MIME type:** {mime_type}",
            f"- **Binary:** {is_binary}",
        ]

        if not is_binary:
            lines.append(f"- **Encoding:** {result.get('encoding', 'unknown')}")
            lines.append(f"- **Lines:** {result.get('lines', 0)}")
            lines.append(f"- **Characters:** {result.get('characters', 0)}")

        lines.append("")

        if is_binary:
            content = result.get("content", "")
            lines.append("## Content (base64)")
            lines.append("")
            # Показываем первые 2000 символов base64
            if len(content) > 2000:
                lines.append(f"```base64\n{content[:2000]}...\n```")
                lines.append(f"\n*[truncated — full length: {len(content)} chars]*")
            else:
                lines.append(f"```base64\n{content}\n```")
        else:
            content = result.get("content", "")
            lines.append("## Content")
            lines.append("")
            # Ограничиваем длину вывода
            if len(content) > 50000:
                lines.append(f"```\n{content[:50000]}\n```\n")
                lines.append(f"*[truncated — total {len(content)} characters]*")
            else:
                lines.append(f"```\n{content}\n```")

        return "\n".join(lines)

    def _format_write_result(self, operation: str, result: dict[str, Any]) -> str:
        """Форматировать результат write/append."""
        return (
            f"# {operation.title()} Complete\n\n"
            f"- **Path:** {result.get('path', '')}\n"
            f"- **Bytes written:** {result.get('bytes_written_human', '')}\n"
            f"- **Encoding:** {result.get('encoding', '')}\n"
            f"- **Final size:** {result.get('final_size_human', '')}\n"
        )

    def _format_delete_result(self, result: dict[str, Any]) -> str:
        """Форматировать результат delete."""
        was_dir = result.get("was_directory", False)
        kind = "directory" if was_dir else "file"
        return (
            f"# Deleted {kind}\n\n"
            f"- **Path:** {result.get('path', '')}\n"
            f"- **Size:** {result.get('deleted_size_human', '')}\n"
        )

    def _format_move_copy_result(
        self, operation: str, result: dict[str, Any]
    ) -> str:
        """Форматировать результат move/copy."""
        return (
            f"# {operation.title()} Complete\n\n"
            f"- **Source:** {result.get('source', '')}\n"
            f"- **Destination:** {result.get('destination', '')}\n"
            f"- **Size:** {result.get('size_human', '')}\n"
        )

    def _format_list_result(self, result: dict[str, Any]) -> str:
        """Форматировать результат list."""
        lines: list[str] = [
            f"# Directory: {result.get('path', '')}",
            "",
        ]

        directories = result.get("directories", [])
        files = result.get("files", [])
        truncated = result.get("truncated", False)

        if directories:
            lines.append(f"## Directories ({len(directories)})")
            lines.append("")
            for d in directories[:100]:  # Ограничиваем вывод
                lines.append(f"- 📁 {d['name']}/")
            if len(directories) > 100:
                lines.append(f"- ... and {len(directories) - 100} more")
            lines.append("")

        if files:
            lines.append(f"## Files ({len(files)})")
            lines.append("")
            for f in files[:100]:
                size = f.get("size_human", "?")
                lines.append(f"- 📄 {f['name']} ({size})")
            if len(files) > 100:
                lines.append(f"- ... and {len(files) - 100} more")
            lines.append("")

        if truncated:
            lines.append(
                f"*[truncated — total items: {result.get('total_items', '?')}]*"
            )

        if not directories and not files:
            lines.append("*(empty directory)*")

        return "\n".join(lines)

    def _format_tree_result(self, result: dict[str, Any]) -> str:
        """Форматировать результат tree."""
        lines = [
            f"# Tree: {result.get('path', '')}",
            "",
            f"Max depth: {result.get('max_depth', '?')}",
            f"Items: {result.get('items_count', 0)}",
            "",
            "```",
            result.get("tree", ""),
            "```",
        ]
        if result.get("truncated"):
            lines.append("\n*[truncated]*")
        return "\n".join(lines)

    def _format_dir_op_result(self, operation: str, result: dict[str, Any]) -> str:
        """Форматировать результат mkdir/rmdir."""
        path = result.get("path", "")
        if operation == "mkdir":
            if result.get("already_exists"):
                return f"# Directory already exists: {path}"
            return f"# Created directory: {path}"
        else:  # rmdir
            return f"# Removed directory: {path}"

    def _format_stat_result(self, result: dict[str, Any]) -> str:
        """Форматировать результат stat."""
        kind = "directory" if result.get("is_directory") else "file"
        if result.get("is_symlink"):
            kind += " (symlink)"

        lines = [
            f"# {kind.title()}: {result.get('path', '')}",
            "",
            f"- **Name:** {result.get('name', '')}",
            f"- **Size:** {result.get('size_human', '')}",
            f"- **Permissions:** {result.get('permissions', '')}",
            f"- **MIME type:** {result.get('mime_type') or 'N/A'}",
            f"- **Extension:** {result.get('extension') or 'N/A'}",
            f"- **Modified:** {result.get('modified_at') or 'N/A'}",
            f"- **Accessed:** {result.get('accessed_at') or 'N/A'}",
            f"- **Created:** {result.get('created_at') or 'N/A'}",
        ]
        return "\n".join(lines)

    def _format_exists_result(self, result: dict[str, Any]) -> str:
        """Форматировать результат exists."""
        exists = result.get("exists", False)
        path = result.get("path", "")
        if not exists:
            return f"# Path does not exist: {path}"

        parts = []
        if result.get("is_directory"):
            parts.append("directory")
        if result.get("is_file"):
            parts.append("file")
        if result.get("is_symlink"):
            parts.append("symlink")

        return f"# Path exists: {path} ({', '.join(parts) if parts else 'unknown type'})"

    def _format_search_result(self, result: dict[str, Any]) -> str:
        """Форматировать результат search."""
        matches = result.get("matches", [])
        lines = [
            f"# Search Results",
            "",
            f"- **Pattern:** {result.get('pattern', '')}",
            f"- **In directory:** {result.get('path', '')}",
            f"- **Recursive:** {result.get('recursive', False)}",
            f"- **Matches:** {len(matches)}",
            "",
        ]

        if matches:
            for m in matches[:50]:
                kind = "📁" if m.get("is_directory") else "📄"
                size = m.get("size_human", "?")
                lines.append(f"- {kind} {m['path']} ({size})")
            if len(matches) > 50:
                lines.append(f"- ... and {len(matches) - 50} more")

        if result.get("truncated"):
            lines.append(
                f"\n*[truncated — max {result.get('matches_count', '?')} shown]*"
            )

        if not matches:
            lines.append("*(no matches found)*")

        return "\n".join(lines)


# ============================================================================
# Self-test
# ============================================================================


if __name__ == "__main__":
    import asyncio
    import tempfile

    async def _self_test() -> None:
        """Проверка корректности file_ops tool."""
        # Создаём temp sandbox
        with tempfile.TemporaryDirectory() as sandbox:
            sandbox_path = Path(sandbox)

            # Заполняем тестовыми данными
            (sandbox_path / "test.txt").write_text("Hello, World!", encoding="utf-8")
            (sandbox_path / "subdir").mkdir()
            (sandbox_path / "subdir" / "nested.md").write_text("# Nested", encoding="utf-8")
            (sandbox_path / ".hidden").write_text("hidden", encoding="utf-8")

            config = FileOpsConfig(base_dir=sandbox)
            tool = FileOpsTool(config)

            print(f"📁 FileOps Tool self-test (sandbox: {sandbox})\n")

            passed = 0
            failed = 0

            # === read ===
            print("=== read ===")
            result = await tool.execute(operation="read", path="test.txt")
            is_ok = result.is_success() and "Hello, World!" in result.get_text()
            print(f"  {'✅' if is_ok else '❌'} read text file")
            (passed if is_ok else (failed := failed + 1)) or (passed := passed + 1) if is_ok else None
            if is_ok:
                passed += 1
            else:
                failed += 1

            # === exists ===
            print("\n=== exists ===")
            result = await tool.execute(operation="exists", path="test.txt")
            is_ok = result.is_success() and result.metadata and result.metadata.get("exists") is True
            print(f"  {'✅' if is_ok else '❌'} exists (file)")
            if is_ok:
                passed += 1
            else:
                failed += 1

            result = await tool.execute(operation="exists", path="nonexistent.txt")
            is_ok = result.is_success() and result.metadata and result.metadata.get("exists") is False
            print(f"  {'✅' if is_ok else '❌'} exists (not found)")
            if is_ok:
                passed += 1
            else:
                failed += 1

            # === stat ===
            print("\n=== stat ===")
            result = await tool.execute(operation="stat", path="test.txt")
            is_ok = result.is_success() and result.metadata and result.metadata.get("is_file") is True
            print(f"  {'✅' if is_ok else '❌'} stat file")
            if is_ok:
                passed += 1
            else:
                failed += 1

            # === list ===
            print("\n=== list ===")
            result = await tool.execute(operation="list", path=".")
            is_ok = result.is_success() and result.metadata and len(result.metadata.get("files", [])) >= 1
            print(f"  {'✅' if is_ok else '❌'} list root")
            if is_ok:
                passed += 1
            else:
                failed += 1

            # === tree ===
            print("\n=== tree ===")
            result = await tool.execute(operation="tree", path=".", max_depth=2)
            is_ok = result.is_success() and "subdir" in result.get_text()
            print(f"  {'✅' if is_ok else '❌'} tree")
            if is_ok:
                passed += 1
            else:
                failed += 1

            # === search ===
            print("\n=== search ===")
            result = await tool.execute(
                operation="search", path=".", pattern="*.txt", recursive=True
            )
            is_ok = result.is_success() and result.metadata and len(result.metadata.get("matches", [])) >= 1
            print(f"  {'✅' if is_ok else '❌'} search *.txt")
            if is_ok:
                passed += 1
            else:
                failed += 1

            # === write ===
            print("\n=== write ===")
            result = await tool.execute(
                operation="write",
                path="output/new.txt",
                content="New content",
            )
            is_ok = result.is_success() and (sandbox_path / "output" / "new.txt").exists()
            print(f"  {'✅' if is_ok else '❌'} write with auto mkdir")
            if is_ok:
                passed += 1
            else:
                failed += 1

            # === append ===
            print("\n=== append ===")
            result = await tool.execute(
                operation="append",
                path="output/new.txt",
                content="\nAppended line",
            )
            final_content = (sandbox_path / "output" / "new.txt").read_text(encoding="utf-8")
            is_ok = result.is_success() and "Appended line" in final_content
            print(f"  {'✅' if is_ok else '❌'} append")
            if is_ok:
                passed += 1
            else:
                failed += 1

            # === copy ===
            print("\n=== copy ===")
            result = await tool.execute(
                operation="copy",
                path="test.txt",
                destination="test_copy.txt",
            )
            is_ok = result.is_success() and (sandbox_path / "test_copy.txt").exists()
            print(f"  {'✅' if is_ok else '❌'} copy")
            if is_ok:
                passed += 1
            else:
                failed += 1

            # === move ===
            print("\n=== move ===")
            result = await tool.execute(
                operation="move",
                path="test_copy.txt",
                destination="test_moved.txt",
            )
            is_ok = (
                result.is_success()
                and (sandbox_path / "test_moved.txt").exists()
                and not (sandbox_path / "test_copy.txt").exists()
            )
            print(f"  {'✅' if is_ok else '❌'} move")
            if is_ok:
                passed += 1
            else:
                failed += 1

            # === mkdir ===
            print("\n=== mkdir ===")
            result = await tool.execute(
                operation="mkdir", path="new/deeply/nested/dir"
            )
            is_ok = result.is_success() and (sandbox_path / "new" / "deeply" / "nested" / "dir").is_dir()
            print(f"  {'✅' if is_ok else '❌'} mkdir (recursive)")
            if is_ok:
                passed += 1
            else:
                failed += 1

            # === rmdir ===
            print("\n=== rmdir ===")
            result = await tool.execute(
                operation="rmdir", path="new/deeply/nested/dir"
            )
            is_ok = result.is_success() and not (sandbox_path / "new" / "deeply" / "nested" / "dir").exists()
            print(f"  {'✅' if is_ok else '❌'} rmdir")
            if is_ok:
                passed += 1
            else:
                failed += 1

            # === delete ===
            print("\n=== delete ===")
            result = await tool.execute(operation="delete", path="test_moved.txt")
            is_ok = result.is_success() and not (sandbox_path / "test_moved.txt").exists()
            print(f"  {'✅' if is_ok else '❌'} delete file")
            if is_ok:
                passed += 1
            else:
                failed += 1

            # === Security: path traversal ===
            print("\n=== Security tests ===")
            result = await tool.execute(operation="read", path="../../../etc/passwd")
            is_ok = result.is_failure()
            print(f"  {'✅' if is_ok else '❌'} path traversal blocked")
            if is_ok:
                passed += 1
            else:
                failed += 1

            result = await tool.execute(operation="read", path="foo/../../bar")
            is_ok = result.is_failure()
            print(f"  {'✅' if is_ok else '❌'} complex traversal blocked")
            if is_ok:
                passed += 1
            else:
                failed += 1

            # === Security: blacklisted paths ===
            result = await tool.execute(operation="read", path=".git/config")
            is_ok = result.is_failure()
            print(f"  {'✅' if is_ok else '❌'} .git blocked")
            if is_ok:
                passed += 1
            else:
                failed += 1

            # === Read-only mode ===
            print("\n=== Read-only mode ===")
            ro_config = FileOpsConfig(base_dir=sandbox, read_only=True)
            ro_tool = FileOpsTool(ro_config)
            result = await ro_tool.execute(
                operation="write", path="forbidden.txt", content="no"
            )
            is_ok = result.is_failure() and "read-only" in result.get_text().lower()
            print(f"  {'✅' if is_ok else '❌'} write blocked in read-only")
            if is_ok:
                passed += 1
            else:
                failed += 1

            # read should still work
            result = await ro_tool.execute(operation="read", path="test.txt")
            is_ok = result.is_success()
            print(f"  {'✅' if is_ok else '❌'} read works in read-only")
            if is_ok:
                passed += 1
            else:
                failed += 1

            print(f"\n📊 Results: {passed} passed, {failed} failed")

    asyncio.run(_self_test())