File Agent：智能文件读写与批量操作

from __future__ import annotations
import asyncio
import hashlib
import mmap
from dataclasses import dataclass, field
from enum import Enum, auto
from pathlib import Path
from typing import AsyncIterator, Optional, Union
import aiofiles
import aiofiles.os


class ReadMode(Enum):
    """读取模式枚举"""
    TEXT = auto()           # 文本模式
    BINARY = auto()         # 二进制模式
    STREAM = auto()         # 流式读取
    MMAP = auto()           # 内存映射模式


@dataclass
class ReadOptions:
    """读取选项配置"""
    mode: ReadMode = ReadMode.TEXT
    encoding: str = "utf-8"
    offset: Optional[int] = None
    size: Optional[int] = None
    chunk_size: int = 8192
    compute_checksum: bool = False
    validate_checksum: Optional[str] = None  # 如果提供，则验证文件校验和


@dataclass
class ReadResult:
    """读取结果封装"""
    content: Union[str, bytes]
    size: int
    checksum: Optional[str] = None
    encoding: Optional[str] = None
    metadata: dict = field(default_factory=dict)
    
    def __post_init__(self):
        if self.checksum is None and isinstance(self.content, (str, bytes)):
            self.checksum = self._compute_checksum()
    
    def _compute_checksum(self) -> str:
        """计算内容校验和"""
        data = self.content.encode() if isinstance(self.content, str) else self.content
        return hashlib.sha256(data).hexdigest()


class FileReadError(Exception):
    """文件读取错误"""
    pass


class ChecksumMismatchError(FileReadError):
    """校验和不匹配错误"""
    pass


class FileAgent:
    """File Agent 核心实现类"""
    
    def __init__(self, base_path: Path, enable_mmap_threshold: int = 10 * 1024 * 1024):
        """
        初始化 File Agent
        
        Args:
            base_path: 允许访问的基础路径（用于安全边界控制）
            enable_mmap_threshold: 启用内存映射的大小阈值（默认10MB）
        """
        self.base_path = base_path.resolve()
        self.enable_mmap_threshold = enable_mmap_threshold
        self._file_locks: dict[str, asyncio.Lock] = {}
        self._read_cache: dict[str, tuple[ReadResult, float]] = {}
        self._cache_ttl = 60.0  # 缓存 TTL 秒数
    
    async def read(self, file_path: Union[str, Path], 
                   options: Optional[ReadOptions] = None) -> ReadResult:
        """
        读取文件内容
        
        Args:
            file_path: 文件路径
            options: 读取选项
            
        Returns:
            ReadResult: 读取结果
            
        Raises:
            FileReadError: 文件读取失败
            ChecksumMismatchError: 校验和验证失败
            PermissionError: 权限不足
        """
        options = options or ReadOptions()
        resolved_path = self._resolve_path(file_path)
        
        # 路径安全检查
        self._validate_path(resolved_path)
        
        # 检查文件是否存在
        if not await aiofiles.os.path.exists(resolved_path):
            raise FileReadError(f"File not found: {resolved_path}")
        
        # 检查是否为常规文件
        stat = await aiofiles.os.stat(resolved_path)
        if not await self._is_regular_file(resolved_path):
            raise FileReadError(f"Not a regular file: {resolved_path}")
        
        # 读取文件内容
        try:
            if options.mode == ReadMode.MMAP:
                content = await self._read_mmap(resolved_path, options)
            elif options.mode == ReadMode.STREAM:
                content = await self._read_stream(resolved_path, options)
            else:
                content = await self._read_simple(resolved_path, options)
        except Exception as e:
            raise FileReadError(f"Failed to read {resolved_path}: {e}")
        
        # 校验和验证
        result = ReadResult(
            content=content,
            size=len(content) if isinstance(content, (str, bytes)) else 0,
            encoding=options.encoding if options.mode == ReadMode.TEXT else None
        )
        
        if options.validate_checksum and result.checksum != options.validate_checksum:
            raise ChecksumMismatchError(
                f"Checksum mismatch: expected {options.validate_checksum}, "
                f"got {result.checksum}"
            )
        
        return result
    
    async def _read_simple(self, path: Path, options: ReadOptions) -> Union[str, bytes]:
        """简单读取模式"""
        mode = "r" if options.mode == ReadMode.TEXT else "rb"
        async with aiofiles.open(path, mode=mode, encoding=options.encoding) as f:
            if options.offset is not None:
                await f.seek(options.offset)
            if options.size is not None:
                return await f.read(options.size)
            return await f.read()
    
    async def _read_stream(self, path: Path, options: ReadOptions) -> str:
        """流式读取模式"""
        chunks = []
        mode = "r" if options.mode == ReadMode.TEXT else "rb"
        async with aiofiles.open(path, mode=mode, encoding=options.encoding) as f:
            while True:
                chunk = await f.read(options.chunk_size)
                if not chunk:
                    break
                chunks.append(chunk)
        return b"".join(chunks) if options.mode == ReadMode.STREAM else "".join(chunks)
    
    async def _read_mmap(self, path: Path, options: ReadOptions) -> bytes:
        """内存映射读取模式（适用于大文件）"""
        file_size = (await aiofiles.os.stat(path)).st_size
        
        if file_size < self.enable_mmap_threshold:
            # 小文件使用普通读取
            return await self._read_simple(path, options)
        
        loop = asyncio.get_event_loop()
        
        def _mmap_read():
            with open(path, "rb") as f:
                with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:
                    offset = options.offset or 0
                    size = options.size or (file_size - offset)
                    return mm[offset:offset + size]
        
        return await loop.run_in_executor(None, _mmap_read)
    
    async def _is_regular_file(self, path: Path) -> bool:
        """检查是否为常规文件"""
        stat = await aiofiles.os.stat(path)
        import stat as stat_module
        return stat_module.S_ISREG(stat.st_mode)
    
    def _resolve_path(self, file_path: Union[str, Path]) -> Path:
        """解析并规范化路径"""
        if isinstance(file_path, str):
            file_path = Path(file_path)
        return (self.base_path / file_path).resolve()
    
    def _validate_path(self, path: Path) -> None:
        """验证路径安全性"""
        # 检查路径前缀，防止路径遍历攻击
        if not str(path).startswith(str(self.base_path)):
            raise SecurityError(
                f"Path traversal attempt detected: {path} is outside base path {self.base_path}"
            )
        
        # 检查敏感文件
        sensitive_patterns = ['.env', '.aws', '.git', '.ssh', 'id_rsa', 'id_ed25519']
        for pattern in sensitive_patterns:
            if pattern in path.parts:
                raise SecurityError(f"Access to sensitive file denied: {path}")
    
    async def read_lines(self, file_path: Union[str, Path], 
                         line_start: Optional[int] = None,
                         line_end: Optional[int] = None) -> AsyncIterator[str]:
        """
        按行迭代读取文件
        
        Args:
            file_path: 文件路径
            line_start: 起始行号（从0开始）
            line_end: 结束行号
            
        Yields:
            str: 文件行内容
        """
        options = ReadOptions(mode=ReadMode.TEXT)
        result = await self.read(file_path, options)
        lines = result.content.split('\n')
        
        start = line_start or 0
        end = line_end or len(lines)
        
        for i in range(start, min(end, len(lines))):
            yield lines[i]

import tempfile
import shutil
from datetime import datetime
from contextlib import asynccontextmanager


@dataclass
class WriteOptions:
    """写入选项配置"""
    mode: str = "w"                    # 写入模式: w=覆盖, a=追加
    encoding: str = "utf-8"
    create_dirs: bool = True           # 自动创建目录
    backup: bool = True                # 写入前创建备份
    atomic: bool = True                # 原子性写入
    fsync: bool = True                 # 强制同步到磁盘
    chmod: Optional[int] = None        # 设置文件权限
    temp_dir: Optional[Path] = None    # 临时文件目录


@dataclass
class WriteResult:
    """写入结果封装"""
    path: Path
    size: int
    checksum: str
    backup_path: Optional[Path] = None
    atomic: bool = False
    metadata: dict = field(default_factory=dict)


class FileWriteError(Exception):
    """文件写入错误"""
    pass


class FileAgent:
    """File Agent - 继续之前的代码"""
    
    async def write(self, file_path: Union[str, Path],
                    content: Union[str, bytes],
                    options: Optional[WriteOptions] = None) -> WriteResult:
        """
        写入文件内容
        
        Args:
            file_path: 文件路径
            content: 文件内容
            options: 写入选项
            
        Returns:
            WriteResult: 写入结果
        """
        options = options or WriteOptions()
        resolved_path = self._resolve_path(file_path)
        
        # 路径安全检查
        self._validate_path(resolved_path)
        
        # 计算校验和
        data = content.encode() if isinstance(content, str) else content
        checksum = hashlib.sha256(data).hexdigest()
        
        backup_path = None
        final_path = resolved_path
        
        try:
            # 确保父目录存在
            if options.create_dirs:
                await self._ensure_parent_dir(resolved_path)
            
            # 备份现有文件
            if options.backup and await aiofiles.os.path.exists(resolved_path):
                backup_path = await self._create_backup(resolved_path)
            
            # 原子性写入
            if options.atomic:
                temp_path = await self._write_temp(resolved_path, data, options)
                final_path = temp_path
            else:
                final_path = await self._write_direct(resolved_path, data, options)
            
            # 设置权限
            if options.chmod is not None:
                await aiofiles.os.chmod(final_path, options.chmod)
            
            return WriteResult(
                path=final_path,
                size=len(data),
                checksum=checksum,
                backup_path=backup_path,
                atomic=options.atomic
            )
            
        except Exception as e:
            # 写入失败，尝试恢复备份
            if backup_path and await aiofiles.os.path.exists(backup_path):
                await self._restore_backup(backup_path, resolved_path)
            raise FileWriteError(f"Failed to write {resolved_path}: {e}")
    
    async def _ensure_parent_dir(self, path: Path) -> None:
        """确保父目录存在"""
        parent = path.parent
        if not await aiofiles.os.path.exists(parent):
            await asyncio.to_thread(lambda: parent.mkdir(parents=True, exist_ok=True))
    
    async def _create_backup(self, path: Path) -> Path:
        """创建文件备份"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        backup_dir = path.parent / ".backups"
        await asyncio.to_thread(lambda: backup_dir.mkdir(parents=True, exist_ok=True))
        backup_path = backup_dir / f"{path.name}.{timestamp}.bak"
        await asyncio.to_thread(lambda: shutil.copy2(path, backup_path))
        return backup_path
    
    async def _write_temp(self, path: Path, data: bytes, 
                          options: WriteOptions) -> Path:
        """原子性写入：先写临时文件，再重命名"""
        temp_dir = options.temp_dir or path.parent
        
        # 创建临时文件
        fd, temp_path_str = tempfile.mkstemp(
            dir=temp_dir,
            prefix=f".{path.name}.",
            suffix=".tmp"
        )
        temp_path = Path(temp_path_str)
        
        try:
            # 写入数据
            loop = asyncio.get_event_loop()
            await loop.run_in_executor(
                None,
                lambda: os.write(fd, data)
            )
            
            # 同步到磁盘
            if options.fsync:
                await loop.run_in_executor(None, lambda: os.fsync(fd))
            
            # 关闭文件描述符
            await loop.run_in_executor(None, lambda: os.close(fd))
            
            # 原子性重命名
            await asyncio.to_thread(
                lambda: shutil.move(str(temp_path), str(path))
            )
            
            return path
            
        except Exception:
            # 清理临时文件
            try:
                await asyncio.to_thread(lambda: os.close(fd))
            except:
                pass
            if temp_path.exists():
                temp_path.unlink()
            raise
    
    async def _write_direct(self, path: Path, data: bytes,
                           options: WriteOptions) -> Path:
        """直接写入模式"""
        mode = "wb" if isinstance(data, bytes) else "w"
        encoding = None if isinstance(data, bytes) else options.encoding
        
        async with aiofiles.open(path, mode=mode, encoding=encoding) as f:
            await f.write(data if isinstance(data, str) else data.decode())
            
            if options.fsync:
                await f.flush()
                await asyncio.to_thread(lambda: os.fsync(f.fileno()))
        
        return path
    
    async def _restore_backup(self, backup_path: Path, original_path: Path) -> None:
        """恢复备份文件"""
        await asyncio.to_thread(
            lambda: shutil.copy2(backup_path, original_path)
        )

@dataclass
class CreateOptions:
    """创建文件/目录选项"""
    is_directory: bool = False
    parents: bool = True                # 创建父目录
    mode: Optional[int] = None          # 权限模式
    exist_ok: bool = True              # 已存在时不报错


@dataclass
class DeleteOptions:
    """删除操作选项"""
    recursive: bool = False             # 递归删除
    backup: bool = True                 # 删除前备份
    unsafe: bool = False                # 强制删除只读文件


@dataclass
class RenameOptions:
    """重命名选项"""
    overwrite: bool = False             # 覆盖目标文件
    backup: bool = True                 # 备份目标文件
    atomic: bool = True                 # 原子性操作


class FileAgent:
    """File Agent - 文件操作原语完整实现"""
    
    async def create(self, file_path: Union[str, Path],
                     options: Optional[CreateOptions] = None) -> Path:
        """
        创建文件或目录
        
        Args:
            file_path: 创建路径
            options: 创建选项
            
        Returns:
            Path: 创建的路径
        """
        options = options or CreateOptions()
        resolved_path = self._resolve_path(file_path)
        self._validate_path(resolved_path)
        
        try:
            if options.is_directory:
                await asyncio.to_thread(
                    lambda: resolved_path.mkdir(
                        parents=options.parents,
                        exist_ok=options.exist_ok,
                        mode=options.mode or 0o755
                    )
                )
            else:
                # 创建文件，确保父目录存在
                if options.parents:
                    await self._ensure_parent_dir(resolved_path)
                await asyncio.to_thread(lambda: resolved_path.touch(mode=options.mode))
            
            return resolved_path
            
        except Exception as e:
            raise FileWriteError(f"Failed to create {resolved_path}: {e}")
    
    async def delete(self, file_path: Union[str, Path],
                     options: Optional[DeleteOptions] = None) -> None:
        """
        删除文件或目录
        
        Args:
            file_path: 删除路径
            options: 删除选项
        """
        options = options or DeleteOptions()
        resolved_path = self._resolve_path(file_path)
        self._validate_path(resolved_path)
        
        if not await aiofiles.os.path.exists(resolved_path):
            if not options.unsafe:  # 非强制模式下不存在则报错
                raise FileReadError(f"Path does not exist: {resolved_path}")
            return
        
        try:
            stat_info = await aiofiles.os.stat(resolved_path)
            is_dir = stat.S_ISDIR(stat_info.st_mode)
            
            # 备份
            if options.backup:
                await self._create_backup(resolved_path)
            
            if is_dir:
                if options.recursive:
                    await asyncio.to_thread(lambda: shutil.rmtree(resolved_path))
                else:
                    await asyncio.to_thread(lambda: resolved_path.rmdir())
            else:
                # 移除只读属性
                if options.unsafe:
                    await asyncio.to_thread(lambda: resolved_path.chmod(0o666))
                await asyncio.to_thread(lambda: resolved_path.unlink())
                
        except Exception as e:
            raise FileWriteError(f"Failed to delete {resolved_path}: {e}")
    
    async def rename(self, src_path: Union[str, Path],
                     dst_path: Union[str, Path],
                     options: Optional[RenameOptions] = None) -> Path:
        """
        重命名或移动文件
        
        Args:
            src_path: 源路径
            dst_path: 目标路径
            options: 重命名选项
            
        Returns:
            Path: 新的路径
        """
        options = options or RenameOptions()
        src_resolved = self._resolve_path(src_path)
        dst_resolved = self._resolve_path(dst_path)
        
        self._validate_path(src_resolved)
        self._validate_path(dst_resolved)
        
        if not await aiofiles.os.path.exists(src_resolved):
            raise FileReadError(f"Source does not exist: {src_resolved}")
        
        if not options.overwrite and await aiofiles.os.path.exists(dst_resolved):
            raise FileWriteError(f"Destination already exists: {dst_resolved}")
        
        # 备份目标
        if options.backup and await aiofiles.os.path.exists(dst_resolved):
            await self._create_backup(dst_resolved)
        
        try:
            if options.atomic:
                # 原子性重命名
                await asyncio.to_thread(
                    lambda: shutil.move(str(src_resolved), str(dst_resolved))
                )
            else:
                await asyncio.to_thread(
                    lambda: src_resolved.rename(dst_resolved)
                )
            
            return dst_resolved
            
        except Exception as e:
            raise FileWriteError(f"Failed to rename {src_resolved} to {dst_resolved}: {e}")

from dataclasses import dataclass, field
from enum import Enum, auto
from typing import Any, Callable, Optional, TypeVar, Generic
from contextlib import asynccontextmanager
import uuid
import json
from pathlib import Path


class TransactionState(Enum):
    """事务状态枚举"""
    PENDING = auto()      # 待执行
    RUNNING = auto()      # 执行中
    COMMITTED = auto()    # 已提交
    ROLLED_BACK = auto()  # 已回滚
    FAILED = auto()       # 失败


@dataclass
class Operation:
    """文件操作封装"""
    id: str
    operation_type: str           # read, write, create, delete, rename
    path: Path
    args: dict
    result: Any = None
    error: Optional[Exception] = None
    undo_data: Any = None         # 用于回滚的数据
    
    def __post_init__(self):
        if not self.id:
            self.id = str(uuid.uuid4())


@dataclass
class Transaction:
    """事务封装"""
    id: str
    state: TransactionState = TransactionState.PENDING
    operations: list[Operation] = field(default_factory=list)
    created_at: datetime = field(default_factory=datetime.now)
    committed_at: Optional[datetime] = None
    
    def __post_init__(self):
        if not self.id:
            self.id = str(uuid.uuid4())


@dataclass
class TransactionOptions:
    """事务选项"""
    auto_rollback: bool = True    # 失败时自动回滚
    max_operations: int = 1000     # 最大操作数
    timeout: Optional[float] = None  # 超时时间（秒）
    checkpoint_interval: int = 10  # 检查点间隔


class TransactionError(Exception):
    """事务执行错误"""
    pass


class TransactionManager:
    """事务管理器"""
    
    def __init__(self, file_agent: 'FileAgent'):
        self.file_agent = file_agent
        self._current_transaction: Optional[Transaction] = None
        self._transaction_log_path: Path = file_agent.base_path / ".transaction_log"
        self._history: list[Transaction] = []
        self._max_history: int = 100
    
    @asynccontextmanager
    async def transaction(self, options: Optional[TransactionOptions] = None):
        """
        事务上下文管理器
        
        Usage:
            async with transaction_manager.transaction() as tx:
                await tx.write("file1.txt", "content1")
                await tx.write("file2.txt", "content2")
        """
        options = options or TransactionOptions()
        tx = Transaction(id=str(uuid.uuid4()))
        self._current_transaction = tx
        
        try:
            yield tx
            await self.commit(tx)
        except Exception as e:
            if options.auto_rollback:
                await self.rollback(tx)
            raise TransactionError(f"Transaction {tx.id} failed: {e}") from e
        finally:
            self._current_transaction = None
    
    async def execute_in_transaction(
        self,
        operations: list[tuple[str, Path, dict]],
        options: Optional[TransactionOptions] = None
    ) -> Transaction:
        """
        在事务中执行一系列操作
        
        Args:
            operations: 操作列表，每个元素为 (operation_type, path, args)
            options: 事务选项
            
        Returns:
            Transaction: 执行结果事务
        """
        options = options or TransactionOptions()
        tx = Transaction(id=str(uuid.uuid4()))
        
        try:
            for op_type, path, args in operations:
                if len(tx.operations) >= options.max_operations:
                    raise TransactionError(f"Too many operations: {options.max_operations}")
                
                op = await self._execute_operation(tx, op_type, path, args)
                tx.operations.append(op)
                
                # 检查点保存
                if len(tx.operations) % options.checkpoint_interval == 0:
                    await self._save_checkpoint(tx)
            
            await self.commit(tx)
            return tx
            
        except Exception as e:
            await self.rollback(tx)
            raise TransactionError(f"Transaction failed: {e}") from e
    
    async def _execute_operation(
        self, tx: Transaction, op_type: str, path: Path, args: dict
    ) -> Operation:
        """执行单个操作并记录"""
        op = Operation(
            id=str(uuid.uuid4()),
            operation_type=op_type,
            path=path,
            args=args
        )
        
        # 准备回滚数据
        if op_type in ('write', 'delete', 'rename'):
            op.undo_data = await self._prepare_undo_data(op_type, path, args)
        
        try:
            tx.state = TransactionState.RUNNING
            
            # 执行操作
            if op_type == 'read':
                op.result = await self.file_agent.read(path, **args)
            elif op_type == 'write':
                op.result = await self.file_agent.write(path, **args)
            elif op_type == 'create':
                op.result = await self.file_agent.create(path, **args)
            elif op_type == 'delete':
                op.result = await self.file_agent.delete(path, **args)
            elif op_type == 'rename':
                op.result = await self.file_agent.rename(**args)
            else:
                raise TransactionError(f"Unknown operation type: {op_type}")
            
            return op
            
        except Exception as e:
            op.error = e
            raise
    
    async def _prepare_undo_data(self, op_type: str, path: Path, args: dict) -> Any:
        """准备回滚所需的数据"""
        if op_type == 'write':
            # 保存原文件内容
            if await aiofiles.os.path.exists(path):
                async with aiofiles.open(path, 'rb') as f:
                    return await f.read()
            return None
            
        elif op_type == 'delete':
            # 保存被删除的文件内容
            async with aiofiles.open(path, 'rb') as f:
                return await f.read()
                
        elif op_type == 'rename':
            # 保存原路径
            return {'src_path': path}
        
        return None
    
    async def commit(self, tx: Transaction) -> None:
        """提交事务"""
        if tx.state in (TransactionState.COMMITTED, TransactionState.ROLLED_BACK):
            raise TransactionError(f"Transaction {tx.id} already {tx.state.value}")
        
        tx.state = TransactionState.COMMITTED
        tx.committed_at = datetime.now()
        
        # 保存到历史
        self._history.append(tx)
        if len(self._history) > self._max_history:
            self._history.pop(0)
        
        # 记录事务日志
        await self._log_transaction(tx)
    
    async def rollback(self, tx: Transaction) -> None:
        """回滚事务"""
        if tx.state in (TransactionState.ROLLED_BACK, TransactionState.COMMITTED):
            return
        
        tx.state = TransactionState.ROLLED_BACK
        
        # 逆序执行撤销操作
        for op in reversed(tx.operations):
            if op.undo_data is None:
                continue
                
            try:
                if op.operation_type == 'write':
                    # 恢复原内容
                    await self.file_agent.write(
                        op.path,
                        op.undo_data,
                        WriteOptions(atomic=False)
                    )
                elif op.operation_type == 'delete':
                    # 恢复被删除的文件
                    await self.file_agent.write(
                        op.path,
                        op.undo_data,
                        WriteOptions(atomic=False)
                    )
                elif op.operation_type == 'rename':
                    # 恢复原路径
                    src_path = op.undo_data['src_path']
                    await self.file_agent.rename(op.path, src_path)
            except Exception as e:
                # 回滚失败，记录错误但继续
                print(f"Rollback failed for operation {op.id}: {e}")
        
        # 记录回滚日志
        await self._log_rollback(tx)
    
    async def _save_checkpoint(self, tx: Transaction) -> None:
        """保存检查点"""
        checkpoint_path = self._transaction_log_path / f"checkpoint_{tx.id}.json"
        await self.file_agent.write(
            checkpoint_path,
            json.dumps({
                'tx_id': tx.id,
                'operations': [
                    {
                        'id': op.id,
                        'type': op.operation_type,
                        'path': str(op.path),
                        'args': op.args
                    }
                    for op in tx.operations
                ]
            }, indent=2),
            WriteOptions()
        )
    
    async def _log_transaction(self, tx: Transaction) -> None:
        """记录事务日志"""
        await self._ensure_log_dir()
        log_path = self._transaction_log_path / f"tx_{tx.id}.log"
        await self.file_agent.write(
            log_path,
            json.dumps({
                'id': tx.id,
                'state': tx.state.value,
                'operations_count': len(tx.operations),
                'created_at': tx.created_at.isoformat(),
                'committed_at': tx.committed_at.isoformat() if tx.committed_at else None
            }, indent=2),
            WriteOptions()
        )
    
    async def _log_rollback(self, tx: Transaction) -> None:
        """记录回滚日志"""
        await self._ensure_log_dir()
        log_path = self._transaction_log_path / f"rollback_{tx.id}.log"
        await self.file_agent.write(
            log_path,
            json.dumps({
                'tx_id': tx.id,
                'rolled_back_at': datetime.now().isoformat(),
                'operations_count': len(tx.operations)
            }, indent=2),
            WriteOptions()
        )
    
    async def _ensure_log_dir(self) -> None:
        """确保日志目录存在"""
        if not await aiofiles.os.path.exists(self._transaction_log_path):
            await asyncio.to_thread(
                lambda: self._transaction_log_path.mkdir(parents=True, exist_ok=True)
            )


class BatchOperation:
    """批量操作封装"""
    
    def __init__(self, file_agent: FileAgent, transaction_manager: TransactionManager):
        self.file_agent = file_agent
        self.tm = transaction_manager
    
    async def batch_write(
        self,
        files: dict[Union[str, Path], Union[str, bytes]],
        options: Optional[TransactionOptions] = None
    ) -> Transaction:
        """
        批量写入文件
        
        Args:
            files: 文件路径到内容的映射
            options: 事务选项
            
        Returns:
            Transaction: 执行结果
        """
        operations = [
            ('write', path, {'content': content, 'options': WriteOptions()})
            for path, content in files.items()
        ]
        
        return await self.tm.execute_in_transaction(operations, options)
    
    async def batch_copy(
        self,
        src_dst_pairs: list[tuple[Union[str, Path], Union[str, Path]]],
        options: Optional[TransactionOptions] = None
    ) -> Transaction:
        """
        批量复制文件
        
        Args:
            src_dst_pairs: 源路径到目标路径的配对列表
            
        Returns:
            Transaction: 执行结果
        """
        operations = []
        
        for src, dst in src_dst_pairs:
            # 先读取源文件
            operations.append(('read', src, {}))
            
        # 执行读取
        tx = Transaction(id=str(uuid.uuid4()))
        results = []
        
        for src, dst in src_dst_pairs:
            content = await self.file_agent.read(src)
            operations.append(('write', dst, {
                'content': content.content,
                'options': WriteOptions()
            }))
        
        return await self.tm.execute_in_transaction(operations, options)
    
    async def atomic_rename_chain(
        self,
        rename_sequence: list[tuple[Union[str, Path], Union[str, Path]]],
        options: Optional[TransactionOptions] = None
    ) -> Transaction:
        """
        原子性批量重命名链
        例如: [('a.txt', 'a.txt.tmp'), ('a.txt.bak', 'a.txt'), ('a.txt.tmp', 'a.txt.bak')]
        
        Args:
            rename_sequence: 重命名序列
            
        Returns:
            Transaction: 执行结果
        """
        operations = [
            ('rename', src, {'dst_path': dst, 'options': RenameOptions()})
            for src, dst in rename_sequence
        ]
        
        return await self.tm.execute_in_transaction(operations, options)

class UndoRedoManager:
    """撤销/重做管理器"""
    
    def __init__(self, max_history: int = 100):
        self._undo_stack: list[Operation] = []
        self._redo_stack: list[Operation] = []
        self._max_history = max_history
    
    def record(self, operation: Operation) -> None:
        """记录操作到撤销栈"""
        self._undo_stack.append(operation)
        self._redo_stack.clear()  # 新操作后清空重做栈
        
        if len(self._undo_stack) > self._max_history:
            self._undo_stack.pop(0)
    
    async def undo(self, file_agent: FileAgent) -> Optional[Operation]:
        """
        撤销上一个操作
        
        Returns:
            Optional[Operation]: 被撤销的操作
        """
        if not self._undo_stack:
            return None
        
        operation = self._undo_stack.pop()
        self._redo_stack.append(operation)
        
        # 执行撤销
        await self._execute_undo(file_agent, operation)
        
        return operation
    
    async def redo(self, file_agent: FileAgent) -> Optional[Operation]:
        """
        重做上一个被撤销的操作
        
        Returns:
            Optional[Operation]: 被重做的操作
        """
        if not self._redo_stack:
            return None
        
        operation = self._redo_stack.pop()
        self._undo_stack.append(operation)
        
        # 执行重做
        await self._execute_redo(file_agent, operation)
        
        return operation
    
    async def _execute_undo(self, file_agent: FileAgent, operation: Operation) -> None:
        """执行撤销"""
        if operation.operation_type == 'write':
            if operation.undo_data is not None:
                await file_agent.write(
                    operation.path,
                    operation.undo_data,
                    WriteOptions(atomic=False)
                )
        elif operation.operation_type == 'create':
            await file_agent.delete(operation.path, DeleteOptions())
        elif operation.operation_type == 'delete':
            await file_agent.write(
                operation.path,
                operation.undo_data,
                WriteOptions()
            )
        elif operation.operation_type == 'rename':
            # 交换源和目标
            original_src = operation.args.get('src_path')
            original_dst = operation.path
            await file_agent.rename(original_dst, original_src)
    
    async def _execute_redo(self, file_agent: FileAgent, operation: Operation) -> None:
        """执行重做（重新执行原操作）"""
        if operation.operation_type == 'write':
            await file_agent.write(operation.path, operation.args.get('content'))
        elif operation.operation_type == 'create':
            await file_agent.create(operation.path, CreateOptions())
        elif operation.operation_type == 'delete':
            await file_agent.delete(operation.path)
        elif operation.operation_type == 'rename':
            await file_agent.rename(**operation.args)
    
    @property
    def can_undo(self) -> bool:
        """是否可以撤销"""
        return len(self._undo_stack) > 0
    
    @property
    def can_redo(self) -> bool:
        """是否可以重做"""
        return len(self._redo_stack) > 0

from enum import Enum, auto
from typing import Optional
import asyncio
import time


class LockType(Enum):
    """锁类型"""
    SHARED = auto()      # 共享锁（读锁）
    EXCLUSIVE = auto()   # 排他锁（写锁）
    RESERVED = auto()    # 预留锁（准备写）


class LockMode(Enum):
    """锁模式"""
    BLOCKING = auto()    # 阻塞等待
    NON_BLOCKING = auto()  # 非阻塞
    TIMEOUT = auto()     # 超时等待


@dataclass
class Lock:
    """锁封装"""
    path: str
    lock_type: LockType
    owner: str
    acquired_at: float
    expires_at: Optional[float] = None


@dataclass
class LockResult:
    """获取锁结果"""
    success: bool
    lock: Optional[Lock] = None
    waited: bool = False
    wait_time: float = 0.0


class FileLockManager:
    """文件锁管理器"""
    
    def __init__(self, timeout: float = 30.0, default_ttl: float = 60.0):
        self.timeout = timeout
        self.default_ttl = default_ttl
        self._locks: dict[str, Lock] = {}
        self._waiters: dict[str, asyncio.Queue] = {}
        self._lock_file_path = Path(".file_locks")
    
    @asynccontextmanager
    async def acquire(
        self,
        path: str,
        lock_type: LockType = LockType.EXCLUSIVE,
        mode: LockMode = LockMode.BLOCKING,
        ttl: Optional[float] = None
    ) -> AsyncIterator[Lock]:
        """
        获取文件锁
        
        Args:
            path: 文件路径
            lock_type: 锁类型
            mode: 获取模式
            ttl: 锁存活时间
            
        Yields:
            Lock: 获取的锁对象
            
        Raises:
            TimeoutError: 获取锁超时
        """
        result = await self._acquire_lock(path, lock_type, mode, ttl)
        
        if not result.success:
            raise TimeoutError(f"Failed to acquire lock for {path}")
        
        try:
            yield result.lock
        finally:
            await self.release(result.lock)
    
    async def _acquire_lock(
        self,
        path: str,
        lock_type: LockType,
        mode: LockMode,
        ttl: Optional[float]
    ) -> LockResult:
        """内部获取锁实现"""
        start_time = time.time()
        ttl = ttl or self.default_ttl
        waited = False
        
        while True:
            lock = await self._try_acquire(path, lock_type, ttl)
            
            if lock is not None:
                return LockResult(
                    success=True,
                    lock=lock,
                    waited=waited,
                    wait_time=time.time() - start_time
                )
            
            if mode == LockMode.NON_BLOCKING:
                return LockResult(success=False)
            
            # 阻塞等待
            waited = True
            wait_start = time.time()
            
            if mode == LockMode.TIMEOUT:
                elapsed = time.time() - start_time
                if elapsed >= self.timeout:
                    return LockResult(success=False)
                remaining = self.timeout - elapsed
                await asyncio.sleep(min(remaining, 0.1))
            else:
                await self._wait_for_unlock(path)
        
        return LockResult(success=False)
    
    async def _try_acquire(self, path: str, lock_type: LockType, ttl: float) -> Optional[Lock]:
        """尝试获取锁"""
        current = self._locks.get(path)
        
        # 检查过期锁
        if current and current.expires_at and time.time() > current.expires_at:
            del self._locks[path]
            current = None
        
        # 无锁或可升级
        if current is None:
            lock = Lock(
                path=path,
                lock_type=lock_type,
                owner=str(id(asyncio.current_task())),
                acquired_at=time.time(),
                expires_at=time.time() + ttl
            )
            self._locks[path] = lock
            return lock
        
        # 共享锁可以升级为排他锁
        if lock_type == LockType.SHARED and current.lock_type == LockType.SHARED:
            return current
        
        # 自己的锁可以升级
        owner_id = str(id(asyncio.current_task()))
        if current.owner == owner_id and lock_type == LockType.EXCLUSIVE:
            # 升级为排他锁
            current.lock_type = LockType.EXCLUSIVE
            current.expires_at = time.time() + ttl
            return current
        
        return None
    
    async def _wait_for_unlock(self, path: str) -> None:
        """等待锁释放"""
        if path not in self._waiters:
            self._waiters[path] = asyncio.Queue()
        
        try:
            await asyncio.wait_for(
                self._waiters[path].get(),
                timeout=1.0
            )
        except asyncio.TimeoutError:
            pass
    
    async def release(self, lock: Lock) -> None:
        """释放锁"""
        if lock.path in self._locks:
            current = self._locks[lock.path]
            if current.owner == str(id(asyncio.current_task())):
                del self._locks[lock.path]
                
                # 通知等待者
                if lock.path in self._waiters:
                    try:
                        self._waiters[lock.path].put_nowait(True)
                    except:
                        pass
    
    async def is_locked(self, path: str) -> bool:
        """检查路径是否被锁定"""
        lock = self._locks.get(path)
        if lock is None:
            return False
        
        if lock.expires_at and time.time() > lock.expires_at:
            del self._locks[path]
            return False
        
        return True
    
    async def get_lock_info(self, path: str) -> Optional[Lock]:
        """获取锁信息"""
        lock = self._locks.get(path)
        if lock and lock.expires_at and time.time() > lock.expires_at:
            del self._locks[path]
            return None
        return lock
    
    async def force_unlock(self, path: str, owner: Optional[str] = None) -> bool:
        """强制解锁"""
        if path not in self._locks:
            return False
        
        current = self._locks[path]
        if owner is None or current.owner == owner:
            del self._locks[path]
            return True
        
        return False

@dataclass
class FileVersion:
    """文件版本信息"""
    version: int
    checksum: str
    modified_at: datetime
    modified_by: str
    content_hash: str


class OptimisticLockManager:
    """乐观锁管理器"""
    
    def __init__(self, file_agent: FileAgent):
        self.file_agent = file_agent
        self._versions: dict[str, FileVersion] = {}
        self._version_file_path = Path(".file_versions")
    
    async def read_with_version(self, path: Path) -> tuple[ReadResult, FileVersion]:
        """
        读取文件并获取版本信息
        
        Returns:
            tuple[ReadResult, FileVersion]: 文件内容和版本信息
        """
        content = await self.file_agent.read(path)
        version = await self._get_or_create_version(path, content.checksum)
        return content, version
    
    async def write_with_check(
        self,
        path: Path,
        content: str | bytes,
        expected_version: int,
        options: Optional[WriteOptions] = None
    ) -> WriteResult:
        """
        写入文件（带版本检查）
        
        Args:
            path: 文件路径
            content: 文件内容
            expected_version: 期望的版本号
            options: 写入选项
            
        Returns:
            WriteResult: 写入结果
            
        Raises:
            VersionConflictError: 版本冲突
        """
        current_version = await self._get_current_version(path)
        
        if current_version and current_version.version != expected_version:
            raise VersionConflictError(
                f"Version conflict: expected {expected_version}, "
                f"current {current_version.version}"
            )
        
        # 执行写入
        result = await self.file_agent.write(path, content, options)
        
        # 更新版本
        new_version = FileVersion(
            version=expected_version + 1,
            checksum=result.checksum,
            modified_at=datetime.now(),
            modified_by=str(id(asyncio.current_task())),
            content_hash=result.checksum
        )
        await self._save_version(path, new_version)
        
        return result
    
    async def _get_current_version(self, path: Path) -> Optional[FileVersion]:
        """获取当前版本"""
        version_file = self._version_file_path / f"{path.name}.v{path.stat().st_ino}"
        if await aiofiles.os.path.exists(version_file):
            data = await self.file_agent.read(version_file)
            return FileVersion(**json.loads(data.content))
        return None
    
    async def _get_or_create_version(self, path: Path, checksum: str) -> FileVersion:
        """获取或创建版本"""
        current = await self._get_current_version(path)
        if current is None:
            current = FileVersion(
                version=1,
                checksum=checksum,
                modified_at=datetime.now(),
                modified_by="system",
                content_hash=checksum
            )
            await self._save_version(path, current)
        return current
    
    async def _save_version(self, path: Path, version: FileVersion) -> None:
        """保存版本信息"""
        version_file = self._version_file_path / f"{path.name}.json"
        await self.file_agent.write(
            version_file,
            json.dumps({
                'version': version.version,
                'checksum': version.checksum,
                'modified_at': version.modified_at.isoformat(),
                'modified_by': version.modified_by,
                'content_hash': version.content_hash
            }, indent=2),
            WriteOptions(create_dirs=True)
        )


class VersionConflictError(Exception):
    """版本冲突错误"""
    pass

from abc import ABC, abstractmethod
from typing import Union


class Operation(ABC):
    """操作基类"""
    
    @abstractmethod
    def apply(self, doc: str) -> str:
        """应用到文档"""
        pass
    
    @abstractmethod
    def transform(self, other: 'Operation') -> 'Operation':
        """与其他操作转换"""
        pass


@dataclass
class InsertOp(Operation):
    """插入操作"""
    position: int
    text: str
    
    def apply(self, doc: str) -> str:
        if self.position > len(doc):
            self.position = len(doc)
        return doc[:self.position] + self.text + doc[self.position:]
    
    def transform(self, other: Operation) -> Operation:
        if isinstance(other, InsertOp):
            return self._transform_insert(other)
        elif isinstance(other, DeleteOp):
            return self._transform_delete(other)
        return self
    
    def _transform_insert(self, other: 'InsertOp') -> InsertOp:
        """与另一个插入操作转换"""
        if other.position <= self.position:
            return InsertOp(self.position + len(other.text), self.text)
        return self
    
    def _transform_delete(self, other: 'DeleteOp') -> InsertOp:
        """与删除操作转换"""
        if other.position >= self.position:
            return self
        elif other.position + other.length <= self.position:
            return InsertOp(self.position - other.length, self.text)
        else:
            # 删除范围包含插入位置，调整
            return InsertOp(other.position, self.text)


@dataclass
class DeleteOp(Operation):
    """删除操作"""
    position: int
    length: int
    
    def apply(self, doc: str) -> str:
        if self.position >= len(doc):
            return doc
        end = min(self.position + self.length, len(doc))
        return doc[:self.position] + doc[end:]
    
    def transform(self, other: Operation) -> Operation:
        if isinstance(other, InsertOp):
            return self._transform_insert(other)
        elif isinstance(other, DeleteOp):
            return self._transform_delete(other)
        return self
    
    def _transform_insert(self, other: 'InsertOp') -> DeleteOp:
        """与插入操作转换"""
        if other.position >= self.position + self.length:
            return self
        elif other.position <= self.position:
            return DeleteOp(self.position + len(other.text), self.length)
        else:
            return DeleteOp(self.position, self.length)
    
    def _transform_delete(self, other: 'DeleteOp') -> DeleteOp:
        """与另一个删除操作转换"""
        # 简化的转换逻辑
        if other.position >= self.position + self.length:
            return self
        elif other.position + other.length <= self.position:
            return DeleteOp(self.position - other.length, self.length)
        elif other.position <= self.position and other.position + other.length >= self.position + self.length:
            # 完全包含，删除自己
            return DeleteOp(other.position, 0)
        else:
            # 部分重叠，需要更复杂的处理
            return DeleteOp(self.position, max(0, self.length - other.length))


class OTEngine:
    """OT 引擎"""
    
    def __init__(self):
        self.pending_ops: list[Operation] = []
        self.sent_ops: list[Operation] = []
    
    def client_transform(self, op: Operation, pending_op: Operation) -> Operation:
        """
        客户端转换：将本地操作与待定操作转换
        
        当本地有一个操作正在等待服务器确认时，新操作需要与它转换
        """
        return op.transform(pending_op)
    
    def server_transform(self, op: Operation, confirmed_op: Operation) -> Operation:
        """
        服务器转换：将操作与已确认操作转换
        
        当服务器收到一个操作时，需要与已广播的操作转换
        """
        return op.transform(confirmed_op)
    
    async def apply_operation(self, doc: str, op: Operation) -> str:
        """应用操作到文档"""
        return op.apply(doc)

@dataclass
class LWWRegister:
    """最后写入 wins (LWW) 寄存器
    
    冲突解决策略：时间戳最大者获胜
    适用场景：配置文件、简单状态
    """
    value: Any
    timestamp: float
    
    def merge(self, other: 'LWWRegister') -> 'LWWRegister':
        if self.timestamp > other.timestamp:
            return self
        return other


@dataclass
class GSet:
    """仅增长集合 (Grow-only Set)
    
    冲突解决策略：集合合并（union）
    限制：只能添加，不能删除
    """
    items: set = field(default_factory=set)
    
    def add(self, item: Any) -> 'GSet':
        return GSet(items=self.items | {item})
    
    def merge(self, other: 'GSet') -> 'GSet':
        return GSet(items=self.items | other.items)
    
    def contains(self, item: Any) -> bool:
        return item in self.items


@dataclass 
class ORSet:
    """可移除集合 (Observed-Remove Set)
    
    支持添加和移除，使用标签区分同名元素
    冲突解决：添加优先于删除
    """
    elements: dict[str, tuple[Any, float]] = field(default_factory=dict)  # tag -> (value, timestamp)
    tombstones: set[str] = field(default_factory=set)  # 已删除的标签
    
    def add(self, item: Any) -> 'ORSet':
        tag = str(uuid.uuid4())
        timestamp = time.time()
        self.elements[tag] = (item, timestamp)
        return self
    
    def remove(self, item: Any) -> 'ORSet':
        timestamp = time.time()
        for tag, (value, _) in self.elements.items():
            if value == item and tag not in self.tombstones:
                self.tombstones.add(tag)
        return self
    
    def merge(self, other: 'ORSet') -> 'ORSet':
        result = ORSet(
            elements={**self.elements, **other.elements},
            tombstones=self.tombstones | other.tombstones
        )
        # 移除在合并过程中发现的 tombstone
        for tag in result.tombstones:
            if tag in result.elements:
                del result.elements[tag]
        return result
    
    def get_items(self) -> set:
        return {v for tag, (v, _) in self.elements.items() if tag not in self.tombstones}


class FileCRDTManager:
    """基于 CRDT 的文件冲突解决管理器"""
    
    def __init__(self):
        self._registers: dict[str, LWWRegister] = {}
        self._sets: dict[str, ORSet] = {}
    
    def resolve_with_lww(
        self,
        path: str,
        local_value: Any,
        remote_value: Any,
        local_timestamp: float,
        remote_timestamp: float
    ) -> Any:
        """使用 LWW 策略解决冲突"""
        local = LWWRegister(value=local_value, timestamp=local_timestamp)
        remote = LWWRegister(value=remote_value, timestamp=remote_timestamp)
        return local.merge(remote).value
    
    def resolve_directory_conflict(
        self,
        local_files: set[str],
        remote_files: set[str]
    ) -> set[str]:
        """
        使用 OR-Set 策略解决目录冲突
        
        目录中的文件添加/删除可以被建模为 OR-Set 操作
        """
        local_set = ORSet()
        remote_set = ORSet()
        
        for f in local_files:
            local_set.add(f)
        for f in remote_files:
            remote_set.add(f)
        
        merged = local_set.merge(remote_set)
        return merged.get_items()

from dataclasses import dataclass
from enum import Enum
from typing import Optional
import mimetypes


class FileCategory(Enum):
    """文件类别"""
    TEXT = "text"
    BINARY = "binary"
    IMAGE = "image"
    AUDIO = "audio"
    VIDEO = "video"
    ARCHIVE = "archive"
    DOCUMENT = "document"
    CODE = "code"
    CONFIG = "config"
    DATA = "data"
    UNKNOWN = "unknown"


@dataclass
class FileTypeInfo:
    """文件类型信息"""
    category: FileCategory
    mime_type: str
    encoding: Optional[str]
    line_ending: Optional[str] = None  # \n, \r\n, \r
    has_bom: bool = False
    bom_type: Optional[str] = None    # utf-8, utf-16, etc.
    is_binary: bool = False
    language: Optional[str] = None     # for code files


# 常见文件魔数
FILE_SIGNATURES = {
    b'\x89PNG\r\n\x1a\n': ('image/png', FileCategory.IMAGE),
    b'\xff\xd8\xff': ('image/jpeg', FileCategory.IMAGE),
    b'GIF87a': ('image/gif', FileCategory.IMAGE),
    b'GIF89a': ('image/gif', FileCategory.IMAGE),
    b'BM': ('image/bmp', FileCategory.IMAGE),
    b'PK\x03\x04': ('application/zip', FileCategory.ARCHIVE),
    b'PK\x05\x06': ('application/zip', FileCategory.ARCHIVE),
    b'\x1f\x8b': ('application/gzip', FileCategory.ARCHIVE),
    b'Rar!': ('application/x-rar', FileCategory.ARCHIVE),
    b'%PDF': ('application/pdf', FileCategory.DOCUMENT),
    b'\x00\x00\x01\x00': ('image/x-icon', FileCategory.IMAGE),
    b'\x00\x00\x02\x00': ('image/x-icon', FileCategory.IMAGE),
    b'\xca\xfe\xba\xbe': ('application/x-java', FileCategory.CODE),
    b'\x7fELF': ('application/x-executable', FileCategory.BINARY),
    b'MZ': ('application/x-executable', FileCategory.BINARY),  # Windows PE
}


class SemanticFileAnalyzer:
    """语义文件分析器"""
    
    def __init__(self):
        self._mime_cache: dict[str, str] = {}
        self._text_extensions = {
            '.txt', '.md', '.rst', '.log', '.csv', '.json', '.yaml', '.yml',
            '.xml', '.html', '.htm', '.css', '.js', '.ts', '.jsx', '.tsx',
            '.py', '.rb', '.java', '.c', '.cpp', '.h', '.hpp', '.go', '.rs',
            '.sh', '.bash', '.zsh', '.ps1', '.bat', '.cmd', '.sql', '.r',
            '.scala', '.swift', '.kt', '.kts', '.php', '.pl', '.pm', '.lua'
        }
        self._config_extensions = {
            '.env', '.ini', '.cfg', '.conf', '.config', '.properties',
            '.toml', '.properties'
        }
    
    async def analyze_file_type(self, path: Path) -> FileTypeInfo:
        """
        分析文件类型
        
        Args:
            path: 文件路径
            
        Returns:
            FileTypeInfo: 文件类型信息
        """
        suffix = path.suffix.lower()
        
        # 首先尝试魔数检测（二进制文件）
        magic_info = await self._detect_by_magic(path)
        if magic_info:
            return magic_info
        
        # 基于扩展名判断
        return await self._detect_by_extension(path, suffix)
    
    async def _detect_by_magic(self, path: Path) -> Optional[FileTypeInfo]:
        """通过魔数检测文件类型"""
        try:
            async with aiofiles.open(path, 'rb') as f:
                header = await f.read(64)
            
            for signature, (mime_type, category) in FILE_SIGNATURES.items():
                if header.startswith(signature):
                    return FileTypeInfo(
                        category=category,
                        mime_type=mime_type,
                        encoding=None,
                        is_binary=True
                    )
        except:
            pass
        
        return None
    
    async def _detect_by_extension(self, path: Path, suffix: str) -> FileTypeInfo:
        """基于扩展名检测"""
        mime_type, _ = mimetypes.guess_type(str(path))
        mime_type = mime_type or 'application/octet-stream'
        
        # 判断类别
        category = self._categorize(suffix)
        
        # 如果是文本文件，检测编码
        encoding = None
        line_ending = None
        has_bom = False
        bom_type = None
        is_binary = False
        language = None
        
        if category in (FileCategory.TEXT, FileCategory.CODE, FileCategory.CONFIG):
            encoding_info = await self._detect_text_encoding(path)
            encoding = encoding_info['encoding']
            line_ending = encoding_info['line_ending']
            has_bom = encoding_info['has_bom']
            bom_type = encoding_info['bom_type']
            language = self._detect_language(suffix)
        else:
            is_binary = True
        
        return FileTypeInfo(
            category=category,
            mime_type=mime_type,
            encoding=encoding,
            line_ending=line_ending,
            has_bom=has_bom,
            bom_type=bom_type,
            is_binary=is_binary,
            language=language
        )
    
    async def _detect_text_encoding(self, path: Path) -> dict:
        """检测文本编码"""
        try:
            async with aiofiles.open(path, 'rb') as f:
                raw = await f.read(4096)
            
            # 检测 BOM
            bom_info = self._detect_bom(raw)
            if bom_info['has_bom']:
                return {
                    'encoding': bom_info['encoding'],
                    'line_ending': self._detect_line_ending(raw),
                    'has_bom': True,
                    'bom_type': bom_info['encoding']
                }
            
            # 尝试解码
            for encoding in ['utf-8', 'utf-8-sigs', 'gbk', 'gb2312', 'gb18030', 
                            'big5', 'shift_jis', 'euc-kr', 'iso-8859-1', 'cp1252']:
                try:
                    raw.decode(encoding)
                    return {
                        'encoding': encoding,
                        'line_ending': self._detect_line_ending(raw),
                        'has_bom': False,
                        'bom_type': None
                    }
                except (UnicodeDecodeError, LookupError):
                    continue
            
            # 默认返回
            return {
                'encoding': 'utf-8',
                'line_ending': '\n',
                'has_bom': False,
                'bom_type': None
            }
            
        except Exception:
            return {
                'encoding': 'utf-8',
                'line_ending': '\n',
                'has_bom': False,
                'bom_type': None
            }
    
    def _detect_bom(self, data: bytes) -> dict:
        """检测 BOM"""
        if data.startswith(b'\xef\xbb\xbf'):
            return {'has_bom': True, 'encoding': 'utf-8-sig'}
        elif data.startswith(b'\xff\xfe'):
            if len(data) >= 4 and data[2:4] == b'\x00\x00':
                return {'has_bom': True, 'encoding': 'utf-32-le'}
            return {'has_bom': True, 'encoding': 'utf-16-le'}
        elif data.startswith(b'\xfe\xff'):
            if len(data) >= 4 and data[2:4] == b'\x00\x00':
                return {'has_bom': True, 'encoding': 'utf-32-be'}
            return {'has_bom': True, 'encoding': 'utf-16-be'}
        elif data.startswith(b'\xff\xfe\x00\x00'):
            return {'has_bom': True, 'encoding': 'utf-32-le'}
        elif data.startswith(b'\x00\x00\xfe\xff'):
            return {'has_bom': True, 'encoding': 'utf-32-be'}
        
        return {'has_bom': False, 'encoding': None}
    
    def _detect_line_ending(self, data: bytes) -> str:
        """检测行尾符"""
        if b'\r\n' in data:
            return '\r\n'
        elif b'\r' in data:
            return '\r'
        elif b'\n' in data:
            return '\n'
        return '\n'
    
    def _categorize(self, suffix: str) -> FileCategory:
        """分类文件"""
        if suffix in self._text_extensions:
            return FileCategory.TEXT
        elif suffix in self._config_extensions:
            return FileCategory.CONFIG
        elif suffix in {'.py', '.js', '.ts', '.java', '.c', '.cpp', '.h', '.hpp',
                        '.go', '.rs', '.rb', '.php', '.swift', '.kt', '.scala'}:
            return FileCategory.CODE
        elif suffix in {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.ico', '.svg', '.webp'}:
            return FileCategory.IMAGE
        elif suffix in {'.mp3', '.wav', '.flac', '.ogg', '.m4a', '.aac'}:
            return FileCategory.AUDIO
        elif suffix in {'.mp4', '.avi', '.mkv', '.mov', '.webm', '.flv'}:
            return FileCategory.VIDEO
        elif suffix in {'.zip', '.tar', '.gz', '.bz2', '.7z', '.rar', '.xz'}:
            return FileCategory.ARCHIVE
        elif suffix in {'.pdf', '.doc', '.docx', '.xls', '.xlsx', '.ppt', '.pptx', '.odt'}:
            return FileCategory.DOCUMENT
        elif suffix in {'.json', '.xml', '.csv', '.tsv', '.sql', '.db', '.sqlite'}:
            return FileCategory.DATA
        elif suffix in {'.exe', '.dll', '.so', '.dylib', '.bin'}:
            return FileCategory.BINARY
        
        return FileCategory.UNKNOWN
    
    def _detect_language(self, suffix: str) -> Optional[str]:
        """检测编程语言"""
        language_map = {
            '.py': 'Python',
            '.js': 'JavaScript',
            '.ts': 'TypeScript',
            '.jsx': 'JavaScript (JSX)',
            '.tsx': 'TypeScript (TSX)',
            '.java': 'Java',
            '.c': 'C',
            '.cpp': 'C++',
            '.h': 'C/C++ Header',
            '.hpp': 'C++ Header',
            '.go': 'Go',
            '.rs': 'Rust',
            '.rb': 'Ruby',
            '.php': 'PHP',
            '.swift': 'Swift',
            '.kt': 'Kotlin',
            '.scala': 'Scala',
            '.lua': 'Lua',
            '.r': 'R',
            '.sql': 'SQL',
            '.sh': 'Shell',
            '.bash': 'Bash',
            '.zsh': 'Zsh',
            '.ps1': 'PowerShell',
            '.lua': 'Lua',
        }
        return language_map.get(suffix)

class IntelligentFileAgent:
    """智能文件代理"""
    
    def __init__(self, file_agent: FileAgent, analyzer: SemanticFileAnalyzer):
        self.file_agent = file_agent
        self.analyzer = analyzer
    
    async def smart_read(self, path: Path) -> tuple[ReadResult, FileTypeInfo]:
        """
        智能读取：根据文件类型自动选择读取方式
        
        Returns:
            tuple[ReadResult, FileTypeInfo]: 文件内容和类型信息
        """
        type_info = await self.analyzer.analyze_file_type(path)
        
        options = ReadOptions(
            mode=ReadMode.BINARY if type_info.is_binary else ReadMode.TEXT,
            encoding=type_info.encoding or 'utf-8'
        )
        
        content = await self.file_agent.read(path, options)
        
        # 文本文件自动转换行尾符
        if not type_info.is_binary and type_info.line_ending != '\n':
            if isinstance(content.content, str):
                content.content = content.content.replace(type_info.line_ending, '\n')
        
        return content, type_info
    
    async def smart_write(
        self,
        path: Path,
        content: str | bytes,
        file_type: Optional[FileTypeInfo] = None,
        preserve_bom: bool = True
    ) -> WriteResult:
        """
        智能写入：根据文件类型自动处理编码和行尾符
        """
        if file_type is None:
            file_type = await self.analyzer.analyze_file_type(path)
        
        # 如果是文本文件
        if isinstance(content, str):
            # 行尾符转换
            if file_type.line_ending and file_type.line_ending != '\n':
                content = content.replace('\n', file_type.line_ending)
            
            options = WriteOptions(
                encoding=file_type.encoding or 'utf-8',
                atomic=True
            )
        else:
            options = WriteOptions(atomic=True)
        
        result = await self.file_agent.write(path, content, options)
        
        # 处理 BOM
        if preserve_bom and not file_type.is_binary and file_type.has_bom:
            await self._ensure_bom(path, file_type.bom_type)
        
        return result
    
    async def _ensure_bom(self, path: Path, bom_type: str) -> None:
        """确保文件有正确的 BOM"""
        bom_bytes = {
            'utf-8-sig': b'\xef\xbb\xbf',
            'utf-16-le': b'\xff\xfe',
            'utf-16-be': b'\xfe\xff',
        }
        
        if bom_type not in bom_bytes:
            return
        
        async with aiofiles.open(path, 'rb') as f:
            existing_bom = await f.read(3)
        
        if not existing_bom.startswith(bom_bytes[bom_type]):
            # 需要添加 BOM
            content = await self.file_agent.read(path)
            await self.file_agent.write(
                path,
                bom_bytes[bom_type] + (content.content if isinstance(content.content, bytes) else content.content.encode()),
                WriteOptions()
            )

import os
import stat as stat_module


class SecurityError(Exception):
    """安全错误异常"""
    pass


class PathTraversalError(SecurityError):
    """路径遍历攻击异常"""
    pass


class SensitiveFileError(SecurityError):
    """敏感文件访问异常"""
    pass


class SecurityValidator:
    """安全验证器"""
    
    # 敏感文件/目录模式
    SENSITIVE_PATTERNS = [
        # 认证凭证
        r'\.env$',
        r'\.env\..+',
        r'\.aws/credentials$',
        r'\.aws/config$',
        r'\.netrc$',
        r'\.git-credentials$',
        r'\.gitlab-ci\.env$',
        
        # SSH 密钥
        r'\.ssh/id_[a-z]+$',
        r'\.ssh/authorized_keys$',
        r'\.ssh/known_hosts$',
        r'\.ssh/config$',
        
        # 私密密钥
        r'\.gnupg/secring\.gpg$',
        r'\.pki/private/.*',
        r'\.npm/_cacache/content-v2/.*',
        r'\.cargo/registry/.*\.tgz$',
        
        # 配置与密码
        r'passwd$',
        r'shadow$',
        r'\.htpasswd$',
        r'passwords?\.txt$',
        r'secrets?\.yml$',
        
        # Git
        r'\.git/config$',
        r'\.git/index$',
        r'\.git/objects/.*',
        
        # 临时文件
        r'\.DS_Store$',
        r'Thumbs\.db$',
        r'desktop\.ini$',
        
        # 系统文件
        r'/etc/passwd$',
        r'/etc/shadow$',
        r'/etc/sudoers$',
    ]
    
    # 禁止访问的路径前缀
    BLOCKED_PATH_PREFIXES = [
        '/proc',
        '/sys',
        '/dev',
        '/boot',
        '/root',
        'C:\\Windows',
        'C:\\Program Files',
        'C:\\Program Files (x86)',
        'C:\\System Volume Information',
    ]
    
    # 允许的文件类型白名单（如果设置）
    ALLOWED_EXTENSIONS: Optional[set] = None
    
    # 禁止的文件类型黑名单
    BLOCKED_EXTENSIONS = {
        '.exe', '.dll', '.so', '.dylib', '.bat', '.cmd', '.ps1', '.vbs',
        '.scr', '.pif', '.msi', '.com', '.jar', '.sh', '.bash'
    }
    
    def __init__(self, base_path: Path, config: Optional[dict] = None):
        self.base_path = base_path.resolve()
        self.config = config or {}
        
        # 从配置更新模式
        if 'sensitive_patterns' in self.config:
            self.SENSITIVE_PATTERNS = self.config['sensitive_patterns']
        if 'allowed_extensions' in self.config:
            self.ALLOWED_EXTENSIONS = set(self.config['allowed_extensions'])
        if 'blocked_extensions' in self.config:
            self.BLOCKED_EXTENSIONS = set(self.config['blocked_extensions'])
    
    def validate_path(self, path: Path, operation: str = 'access') -> None:
        """
        验证路径安全性
        
        Args:
            path: 要验证的路径
            operation: 操作类型 ('read', 'write', 'delete')
            
        Raises:
            PathTraversalError: 检测到路径遍历攻击
            SensitiveFileError: 尝试访问敏感文件
            SecurityError: 其他安全错误
        """
        # 规范化路径
        resolved = self._normalize_path(path)
        
        # 检查路径前缀（防遍历）
        self._check_path_prefix(resolved)
        
        # 检查敏感文件
        self._check_sensitive_file(resolved)
        
        # 检查符号链接
        self._check_symlink(path, resolved)
        
        # 检查设备文件
        self._check_device_file(resolved)
        
        # 检查硬链接
        self._check_hardlink(resolved, operation)
        
        # 检查文件扩展名
        self._check_extension(resolved)
    
    def _normalize_path(self, path: Path) -> Path:
        """规范化路径"""
        # 将路径转换为绝对路径
        if not path.is_absolute():
            path = self.base_path / path
        
        # 解析符号链接
        try:
            resolved = path.resolve()
        except (OSError, RuntimeError):
            # 如果无法解析符号链接，使用绝对路径
            resolved = path.absolute()
        
        return resolved
    
    def _check_path_prefix(self, path: Path) -> None:
        """检查路径前缀，防止遍历攻击"""
        path_str = str(path)
        base_str = str(self.base_path)
        
        # 确保路径在 base_path 下
        if not path_str.startswith(base_str):
            # 进一步检查规范化后的路径
            try:
                path_str = str(path.resolve())
                if not path_str.startswith(base_str):
                    raise PathTraversalError(
                        f"Path traversal attempt detected: {path} resolves to {path_str} "
                        f"which is outside base path {self.base_path}"
                    )
            except (OSError, RuntimeError) as e:
                raise PathTraversalError(f"Cannot resolve path: {path} - {e}")
        
        # 检查是否在禁止访问的路径下
        for blocked in self.BLOCKED_PATH_PREFIXES:
            if path_str.startswith(blocked) or blocked in path_str:
                raise SecurityError(f"Access to blocked path denied: {path}")
    
    def _check_sensitive_file(self, path: Path) -> None:
        """检查敏感文件"""
        import re
        
        path_str = str(path)
        
        for pattern in self.SENSITIVE_PATTERNS:
            if re.search(pattern, path_str, re.IGNORECASE):
                raise SensitiveFileError(
                    f"Access to sensitive file denied: {path} matches pattern {pattern}"
                )
        
        # 额外的精确匹配
        sensitive_names = {
            '.env', '.env.local', '.env.development', '.env.production',
            'id_rsa', 'id_ed25519', 'id_ecdsa', 'id_dsa',
            'authorized_keys', 'known_hosts',
            '.gitconfig', '.gitignore_global', '.git-credentials'
        }
        
        if path.name in sensitive_names:
            raise SensitiveFileError(
                f"Access to sensitive file denied: {path.name} is a sensitive filename"
            )
    
    def _check_symlink(self, original: Path, resolved: Path) -> None:
        """检查符号链接安全性"""
        try:
            if original.is_symlink():
                # 符号链接指向的路径也必须在 base_path 下
                link_target = original.readlink()
                if not link_target.is_absolute():
                    link_target = (original.parent / link_target).resolve()
                else:
                    link_target = link_target.resolve()
                
                link_str = str(link_target)
                if not link_str.startswith(str(self.base_path)):
                    raise SecurityError(
                        f"Symbolic link points outside base path: {original} -> {link_target}"
                    )
        except OSError:
            # 无法检查符号链接（权限问题等），允许访问
            pass
    
    def _check_device_file(self, path: Path) -> None:
        """检查设备文件"""
        try:
            stat_info = path.stat()
            if stat_module.S_ISCHR(stat_info.st_mode) or stat_module.S_ISBLK(stat_info.st_mode):
                raise SecurityError(f"Device file access denied: {path}")
        except OSError:
            pass
    
    def _check_hardlink(self, path: Path, operation: str) -> None:
        """检查硬链接（防止创建指向敏感文件的硬链接）"""
        try:
            stat_info = path.stat()
            # 硬链接数 > 1 表示可能有硬链接存在
            if stat_info.st_nlink > 1 and operation in ('write', 'delete'):
                # 进一步检查是否在 base_path 外有链接
                # 这需要遍历文件系统，开销较大，这里只做简单检查
                pass
        except OSError:
            pass
    
    def _check_extension(self, path: Path) -> None:
        """检查文件扩展名"""
        suffix = path.suffix.lower()
        
        # 黑名单检查
        if suffix in self.BLOCKED_EXTENSIONS:
            raise SecurityError(f"File extension {suffix} is blocked")
        
        # 白名单检查（如果设置了）
        if self.ALLOWED_EXTENSIONS is not None:
            if suffix not in self.ALLOWED_EXTENSIONS:
                raise SecurityError(
                    f"File extension {suffix} is not in the allowed list"
                )


class SandboxFileAgent:
    """沙箱化文件代理"""
    
    def __init__(
        self,
        base_path: Path,
        validator: Optional[SecurityValidator] = None,
        max_file_size: int = 100 * 1024 * 1024,  # 100MB
        max_operation_count: int = 10000
    ):
        self.base_path = base_path
        self.validator = validator or SecurityValidator(base_path)
        self.max_file_size = max_file_size
        self.max_operation_count = max_operation_count
        self._operation_count = 0
    
    async def safe_read(self, path: Path, options: Optional[ReadOptions] = None) -> ReadResult:
        """安全的读取操作"""
        self._check_operation_limit()
        self.validator.validate_path(path, 'read')
        
        # 检查文件大小
        stat = await aiofiles.os.stat(path)
        if stat.st_size > self.max_file_size:
            raise SecurityError(f"File size {stat.st_size} exceeds limit {self.max_file_size}")
        
        return await self._file_agent.read(path, options)
    
    async def safe_write(
        self,
        path: Path,
        content: Union[str, bytes],
        options: Optional[WriteOptions] = None
    ) -> WriteResult:
        """安全的写入操作"""
        self._check_operation_limit()
        self.validator.validate_path(path, 'write')
        
        # 检查写入大小
        data = content.encode() if isinstance(content, str) else content
        if len(data) > self.max_file_size:
            raise SecurityError(f"Write size {len(data)} exceeds limit {self.max_file_size}")
        
        return await self._file_agent.write(path, content, options)
    
    async def safe_delete(self, path: Path, options: Optional[DeleteOptions] = None) -> None:
        """安全的删除操作"""
        self._check_operation_limit()
        self.validator.validate_path(path, 'delete')
        
        return await self._file_agent.delete(path, options)
    
    def _check_operation_count(self) -> None:
        """检查操作计数"""
        self._operation_count += 1
        if self._operation_count > self.max_operation_count:
            raise SecurityError(f"Operation count {self._operation_count} exceeds limit")

"""
完整 File Agent 实现
支持：文件操作原语、事务、撤销/重做、冲突检测、语义感知、安全防护
"""

from __future__ import annotations
import asyncio
import hashlib
import json
import os
import shutil
import tempfile
import time
import uuid
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum, auto
from pathlib import Path
from typing import Any, AsyncIterator, Optional, Union
import aiofiles
import aiofiles.os


# ============== 配置与常量 ==============

class ReadMode(Enum):
    TEXT = auto()
    BINARY = auto()
    STREAM = auto()
    MMAP = auto()


class WriteMode(Enum):
    OVERWRITE = "w"
    APPEND = "a"


# ============== 数据类 ==============

@dataclass
class ReadOptions:
    mode: ReadMode = ReadMode.TEXT
    encoding: str = "utf-8"
    offset: Optional[int] = None
    size: Optional[int] = None
    chunk_size: int = 8192


@dataclass
class WriteOptions:
    mode: WriteMode = WriteMode.OVERWRITE
    encoding: str = "utf-8"
    create_dirs: bool = True
    backup: bool = True
    atomic: bool = True
    fsync: bool = True


@dataclass
class ReadResult:
    content: Union[str, bytes]
    size: int
    checksum: str
    encoding: Optional[str] = None


@dataclass
class WriteResult:
    path: Path
    size: int
    checksum: str
    backup_path: Optional[Path] = None


# ============== 异常类 ==============

class FileAgentError(Exception):
    """File Agent 基础异常"""
    pass


class FileReadError(FileAgentError):
    """文件读取错误"""
    pass


class FileWriteError(FileAgentError):
    """文件写入错误"""
    pass


class SecurityError(FileAgentError):
    """安全错误"""
    pass


class TransactionError(FileAgentError):
    """事务错误"""
    pass


# ============== 核心 File Agent ==============

class FileAgent:
    """
    完整的 File Agent 实现
    
    Features:
    - 文件操作原语（Read/Write/Create/Delete/Rename）
    - 事务性批量操作与回滚
    - 撤销/重做支持
    - 文件锁与乐观锁
    - 语义感知（文件类型、编码检测）
    - 安全边界防护
    """
    
    def __init__(
        self,
        base_path: Union[str, Path],
        enable_undo: bool = True,
        max_undo_stack: int = 100,
        max_file_size: int = 100 * 1024 * 1024
    ):
        self.base_path = Path(base_path).resolve()
        self.enable_undo = enable_undo
        self.max_undo_stack = max_undo_stack
        self.max_file_size = max_file_size
        
        # 内部状态
        self._undo_stack: list[OperationRecord] = []
        self._redo_stack: list[OperationRecord] = []
        self._transaction_stack: list[Transaction] = []
        self._locks: dict[str, LockInfo] = {}
        
        # 确保 base_path 存在
        if not self.base_path.exists():
            self.base_path.mkdir(parents=True, exist_ok=True)
    
    # ============== 读取操作 ==============
    
    async def read(
        self,
        path: Union[str, Path],
        options: Optional[ReadOptions] = None
    ) -> ReadResult:
        """读取文件"""
        options = options or ReadOptions()
        resolved = self._resolve_path(path)
        
        self._validate_path(resolved)
        
        if not await aiofiles.os.path.exists(resolved):
            raise FileReadError(f"File not found: {resolved}")
        
        # 检查文件大小
        stat = await aiofiles.os.stat(resolved)
        if stat.st_size > self.max_file_size:
            raise FileReadError(f"File too large: {stat.st_size} bytes")
        
        # 执行读取
        mode = "r" if options.mode == ReadMode.TEXT else "rb"
        enc = options.encoding if options.mode == ReadMode.TEXT else None
        
        async with aiofiles.open(resolved, mode=mode, encoding=enc) as f:
            if options.offset is not None:
                await f.seek(options.offset)
            
            if options.size is not None:
                content = await f.read(options.size)
            else:
                content = await f.read()
        
        # 计算校验和
        data = content.encode() if isinstance(content, str) else content
        checksum = hashlib.sha256(data).hexdigest()
        
        return ReadResult(
            content=content,
            size=len(data),
            checksum=checksum,
            encoding=enc
        )
    
    async def read_lines(
        self,
        path: Union[str, Path],
        line_start: Optional[int] = None,
        line_end: Optional[int] = None
    ) -> AsyncIterator[str]:
        """按行读取"""
        result = await self.read(path)
        lines = result.content.split('\n') if isinstance(result.content, str) else result.content.decode().split('\n')
        
        start = line_start or 0
        end = line_end or len(lines)
        
        for i in range(start, min(end, len(lines))):
            yield lines[i]
    
    # ============== 写入操作 ==============
    
    async def write(
        self,
        path: Union[str, Path],
        content: Union[str, bytes],
        options: Optional[WriteOptions] = None
    ) -> WriteResult:
        """写入文件"""
        options = options or WriteOptions()
        resolved = self._resolve_path(path)
        
        self._validate_path(resolved)
        
        # 计算数据
        data = content.encode() if isinstance(content, str) else content
        checksum = hashlib.sha256(data).hexdigest()
        
        backup_path = None
        
        # 确保目录存在
        if options.create_dirs:
            await self._ensure_parent_dir(resolved)
        
        # 备份现有文件
        if options.backup and await aiofiles.os.path.exists(resolved):
            backup_path = await self._create_backup(resolved)
        
        # 写入
        if options.atomic:
            temp_path = await self._write_atomic(resolved, data, options)
        else:
            temp_path = await self._write_direct(resolved, data, options)
        
        # 记录撤销信息
        if self.enable_undo:
            self._record_operation(
                OperationRecord(
                    op_type='write',
                    path=resolved,
                    undo_data=backup_path.read_bytes() if backup_path and backup_path.exists() else None
                )
            )
        
        return WriteResult(
            path=temp_path,
            size=len(data),
            checksum=checksum,
            backup_path=backup_path
        )
    
    async def _write_atomic(
        self,
        path: Path,
        data: bytes,
        options: WriteOptions
    ) -> Path:
        """原子性写入"""
        fd, temp_path_str = tempfile.mkstemp(
            dir=path.parent,
            prefix=f".{path.name}.",
            suffix=".tmp"
        )
        temp_path = Path(temp_path_str)
        
        try:
            loop = asyncio.get_event_loop()
            await loop.run_in_executor(None, lambda: os.write(fd, data))
            
            if options.fsync:
                await loop.run_in_executor(None, lambda: os.fsync(fd))
            
            await loop.run_in_executor(None, lambda: os.close(fd))
            
            # 原子性重命名
            await asyncio.to_thread(
                lambda: shutil.move(str(temp_path), str(path))
            )
            
            return path
            
        except Exception:
            try:
                await asyncio.to_thread(lambda: os.close(fd))
            except:
                pass
            if temp_path.exists():
                temp_path.unlink()
            raise
    
    async def _write_direct(
        self,
        path: Path,
        data: bytes,
        options: WriteOptions
    ) -> Path:
        """直接写入"""
        mode = "wb" if isinstance(data, bytes) else "w"
        enc = None if isinstance(data, bytes) else options.encoding
        
        async with aiofiles.open(path, mode=mode, encoding=enc) as f:
            await f.write(data if isinstance(data, str) else data.decode())
            
            if options.fsync:
                await f.flush()
                await asyncio.to_thread(lambda: os.fsync(f.fileno()))
        
        return path
    
    # ============== 文件操作 ==============
    
    async def create(
        self,
        path: Union[str, Path],
        is_dir: bool = False,
        parents: bool = True,
        mode: Optional[int] = None
    ) -> Path:
        """创建文件或目录"""
        resolved = self._resolve_path(path)
        self._validate_path(resolved)
        
        if is_dir:
            await asyncio.to_thread(
                lambda: resolved.mkdir(parents=parents, exist_ok=True, mode=mode or 0o755)
            )
        else:
            if parents:
                await self._ensure_parent_dir(resolved)
            await asyncio.to_thread(lambda: resolved.touch(mode=mode))
        
        if self.enable_undo:
            self._record_operation(
                OperationRecord(op_type='create', path=resolved, undo_data=None)
            )
        
        return resolved
    
    async def delete(
        self,
        path: Union[str, Path],
        recursive: bool = False,
        backup: bool = True
    ) -> None:
        """删除文件或目录"""
        resolved = self._resolve_path(path)
        self._validate_path(resolved)
        
        if not await aiofiles.os.path.exists(resolved):
            return
        
        # 保存删除数据用于撤销
        undo_data = None
        if self.enable_undo and backup:
            stat = await aiofiles.os.stat(resolved)
            is_dir = stat.st_mode & 0o170000 == 0o40000
            if is_dir:
                undo_data = {'type': 'dir', 'path': str(resolved)}
            else:
                async with aiofiles.open(resolved, 'rb') as f:
                    undo_data = {'type': 'file', 'data': await f.read()}
        
        # 执行删除
        stat = await aiofiles.os.stat(resolved)
        is_dir = stat.st_mode & 0o170000 == 0o40000
        
        if is_dir:
            if recursive:
                await asyncio.to_thread(lambda: shutil.rmtree(resolved))
            else:
                await asyncio.to_thread(lambda: resolved.rmdir())
        else:
            await asyncio.to_thread(lambda: resolved.unlink())
        
        if self.enable_undo:
            self._record_operation(
                OperationRecord(op_type='delete', path=resolved, undo_data=undo_data)
            )
    
    async def rename(
        self,
        src: Union[str, Path],
        dst: Union[str, Path],
        overwrite: bool = False
    ) -> Path:
        """重命名或移动"""
        src_resolved = self._resolve_path(src)
        dst_resolved = self._resolve_path(dst)
        
        self._validate_path(src_resolved)
        self._validate_path(dst_resolved)
        
        if not await aiofiles.os.path.exists(src_resolved):
            raise FileWriteError(f"Source not found: {src_resolved}")
        
        if await aiofiles.os.path.exists(dst_resolved) and not overwrite:
            raise FileWriteError(f"Destination exists: {dst_resolved}")
        
        await asyncio.to_thread(
            lambda: shutil.move(str(src_resolved), str(dst_resolved))
        )
        
        if self.enable_undo:
            self._record_operation(
                OperationRecord(
                    op_type='rename',
                    path=dst_resolved,
                    undo_data={'src': str(src_resolved), 'dst': str(dst_resolved)}
                )
            )
        
        return dst_resolved
    
    # ============== 撤销/重做 ==============
    
    async def undo(self) -> Optional[str]:
        """撤销上一个操作"""
        if not self._undo_stack:
            return None
        
        op = self._undo_stack.pop()
        self._redo_stack.append(op)
        
        await self._execute_undo(op)
        return op.op_type
    
    async def redo(self) -> Optional[str]:
        """重做上一个被撤销的操作"""
        if not self._redo_stack:
            return None
        
        op = self._redo_stack.pop()
        self._undo_stack.append(op)
        
        await self._execute_redo(op)
        return op.op_type
    
    async def _execute_undo(self, op: OperationRecord) -> None:
        """执行撤销"""
        if op.op_type == 'write':
            if op.undo_data and isinstance(op.undo_data, bytes):
                await self.write(op.path, op.undo_data, WriteOptions(atomic=False))
        elif op.op_type == 'delete':
            if op.undo_data:
                if op.undo_data.get('type') == 'dir':
                    await self.create(op.path, is_dir=True)
                else:
                    await self.write(op.path, op.undo_data['data'], WriteOptions())
        elif op.op_type == 'rename':
            info = op.undo_data
            await self.rename(info['dst'], info['src'])
        elif op.op_type == 'create':
            await self.delete(op.path)
    
    async def _execute_redo(self, op: OperationRecord) -> None:
        """执行重做"""
        if op.op_type == 'write':
            # 重新执行写入需要原始内容，这需要记录
            pass
        elif op.op_type == 'delete':
            await self.delete(op.path)
        elif op.op_type == 'rename':
            info = op.undo_data
            await self.rename(info['src'], info['dst'])
        elif op.op_type == 'create':
            await self.create(op.path)
    
    def _record_operation(self, op: OperationRecord) -> None:
        """记录操作"""
        self._undo_stack.append(op)
        self._redo_stack.clear()
        
        if len(self._undo_stack) > self.max_undo_stack:
            self._undo_stack.pop(0)
    
    # ============== 事务支持 ==============
    
    @asynccontextmanager
    async def transaction(self):
        """事务上下文管理器"""
        tx = Transaction(id=str(uuid.uuid4()))
        self._transaction_stack.append(tx)
        
        try:
            yield tx
            await self._commit_transaction(tx)
        except Exception as e:
            await self._rollback_transaction(tx)
            raise TransactionError(f"Transaction failed: {e}") from e
        finally:
            if tx in self._transaction_stack:
                self._transaction_stack.remove(tx)
    
    async def _commit_transaction(self, tx: Transaction) -> None:
        """提交事务"""
        tx.committed_at = datetime.now()
        tx.state = TransactionState.COMMITTED
    
    async def _rollback_transaction(self, tx: Transaction) -> None:
        """回滚事务"""
        tx.state = TransactionState.ROLLED_BACK
        # 逆序撤销操作
        for op in reversed(tx.operations):
            await self._execute_undo(op)
    
    # ============== 辅助方法 ==============
    
    def _resolve_path(self, path: Union[str, Path]) -> Path:
        """解析路径"""
        if isinstance(path, str):
            path = Path(path)
        if not path.is_absolute():
            path = self.base_path / path
        return path.resolve()
    
    def _validate_path(self, path: Path) -> None:
        """验证路径安全性"""
        path_str = str(path)
        base_str = str(self.base_path)
        
        if not path_str.startswith(base_str):
            raise SecurityError(f"Path outside base: {path}")
        
        # 检查敏感文件
        sensitive = {'.env', '.ssh', '.git', 'id_rsa', 'id_ed25519', 'authorized_keys'}
        for part in path.parts:
            if part in sensitive:
                raise SecurityError(f"Access to sensitive path denied: {path}")
    
    async def _ensure_parent_dir(self, path: Path) -> None:
        """确保父目录存在"""
        parent = path.parent
        if not await aiofiles.os.path.exists(parent):
            await asyncio.to_thread(lambda: parent.mkdir(parents=True, exist_ok=True))
    
    async def _create_backup(self, path: Path) -> Path:
        """创建备份"""
        backup_dir = path.parent / ".backups"
        await asyncio.to_thread(lambda: backup_dir.mkdir(parents=True, exist_ok=True))
        
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        backup_path = backup_dir / f"{path.name}.{timestamp}.bak"
        
        await asyncio.to_thread(lambda: shutil.copy2(path, backup_path))
        return backup_path


# ============== 辅助数据类 ==============

@dataclass
class OperationRecord:
    """操作记录"""
    op_type: str
    path: Path
    undo_data: Any
    timestamp: datetime = field(default_factory=datetime.now)


class TransactionState(Enum):
    PENDING = auto()
    COMMITTED = auto()
    ROLLED_BACK = auto()


@dataclass
class Transaction:
    id: str
    operations: list[OperationRecord] = field(default_factory=list)
    state: TransactionState = TransactionState.PENDING
    committed_at: Optional[datetime] = None


@dataclass
class LockInfo:
    owner: str
    acquired_at: float
    expires_at: Optional[float] = None


# ============== 使用示例 ==============

async def main():
    """File Agent 使用示例"""
    
    # 创建 File Agent
    agent = FileAgent("./workspace", enable_undo=True)
    
    # 基本读写
    result = await agent.write("test.txt", "Hello, File Agent!")
    print(f"Written: {result.path}, checksum: {result.checksum}")
    
    content = await agent.read("test.txt")
    print(f"Read: {content.content}, size: {content.size}")
    
    # 批量操作
    files = {
        "file1.txt": "Content 1",
        "file2.txt": "Content 2",
        "file3.txt": "Content 3"
    }
    
    async with agent.transaction():
        for path, content in files.items():
            await agent.write(path, content)
        print("Batch write completed")
    
    # 撤销操作
    await agent.undo()
    print("Undo completed")
    
    await agent.redo()
    print("Redo completed")


if __name__ == "__main__":
    asyncio.run(main())

import pytest
from pathlib import Path
import tempfile
import shutil


class TestFileAgent:
    """File Agent 测试套件"""
    
    @pytest.fixture
    async def agent(self):
        """创建测试用 File Agent"""
        temp_dir = tempfile.mkdtemp()
        yield FileAgent(temp_dir)
        shutil.rmtree(temp_dir, ignore_errors=True)
    
    # ============== 基础操作测试 ==============
    
    @pytest.mark.asyncio
    async def test_write_and_read(self, agent):
        """测试基本读写"""
        content = "Hello, File Agent!"
        await agent.write("test.txt", content)
        
        result = await agent.read("test.txt")
        assert result.content == content
        assert result.size == len(content)
    
    @pytest.mark.asyncio
    async def test_read_nonexistent(self, agent):
        """测试读取不存在的文件"""
        with pytest.raises(FileReadError):
            await agent.read("nonexistent.txt")
    
    @pytest.mark.asyncio
    async def test_write_binary(self, agent):
        """测试二进制写入"""
        data = bytes(range(256))
        await agent.write("binary.bin", data)
        
        result = await agent.read("binary.bin")
        assert result.content == data
    
    # ============== 撤销/重做测试 ==============
    
    @pytest.mark.asyncio
    async def test_undo_write(self, agent):
        """测试撤销写入"""
        await agent.write("test.txt", "Original")
        await agent.undo()
        
        exists = await aiofiles.os.path.exists(agent.base_path / "test.txt")
        assert not exists
    
    @pytest.mark.asyncio
    async def test_undo_delete(self, agent):
        """测试撤销删除"""
        await agent.write("test.txt", "Content")
        await agent.delete("test.txt")
        await agent.undo()
        
        result = await agent.read("test.txt")
        assert result.content == "Content"
    
    @pytest.mark.asyncio
    async def test_undo_rename(self, agent):
        """测试撤销重命名"""
        await agent.write("original.txt", "Content")
        await agent.rename("original.txt", "renamed.txt")
        await agent.undo()
        
        exists_original = await aiofiles.os.path.exists(agent.base_path / "original.txt")
        exists_renamed = await aiofiles.os.path.exists(agent.base_path / "renamed.txt")
        assert exists_original
        assert not exists_renamed
    
    @pytest.mark.asyncio
    async def test_redo(self, agent):
        """测试重做"""
        await agent.write("test.txt", "Content")
        await agent.undo()
        await agent.redo()
        
        result = await agent.read("test.txt")
        assert result.content == "Content"
    
    # ============== 事务测试 ==============
    
    @pytest.mark.asyncio
    async def test_transaction_commit(self, agent):
        """测试事务提交"""
        async with agent.transaction():
            await agent.write("file1.txt", "Content 1")
            await agent.write("file2.txt", "Content 2")
        
        result1 = await agent.read("file1.txt")
        result2 = await agent.read("file2.txt")
        assert result1.content == "Content 1"
        assert result2.content == "Content 2"
    
    @pytest.mark.asyncio
    async def test_transaction_rollback(self, agent):
        """测试事务回滚"""
        try:
            async with agent.transaction():
                await agent.write("file1.txt", "Content 1")
                await agent.write("file2.txt", "Content 2")
                raise ValueError("Simulated error")
        except:
            pass
        
        exists1 = await aiofiles.os.path.exists(agent.base_path / "file1.txt")
        exists2 = await aiofiles.os.path.exists(agent.base_path / "file2.txt")
        assert not exists1
        assert not exists2
    
    # ============== 安全测试 ==============
    
    @pytest.mark.asyncio
    async def test_path_traversal_blocked(self, agent):
        """测试路径遍历防护"""
        with pytest.raises(SecurityError):
            await agent.read("../etc/passwd")
    
    @pytest.mark.asyncio
    async def test_sensitive_file_blocked(self, agent):
        """测试敏感文件防护"""
        with pytest.raises(SecurityError):
            await agent.read(".env")
    
    @pytest.mark.asyncio
    async def test_absolute_path_restricted(self, agent):
        """测试绝对路径限制"""
        with pytest.raises(SecurityError):
            await agent.read("/etc/passwd")

"""
File Agent - 智能文件读写与批量操作
作者：HOS(安全风信子)
版本：1.0.0
"""

from __future__ import annotations
import asyncio
import hashlib
import json
import os
import shutil
import tempfile
import time
import uuid
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum, auto
from pathlib import Path
from typing import Any, AsyncIterator, Optional, Union
import aiofiles
import aiofiles.os


# ============== 枚举定义 ==============

class ReadMode(Enum):
    """读取模式"""
    TEXT = auto()
    BINARY = auto()
    STREAM = auto()
    MMAP = auto()


class WriteMode(Enum):
    """写入模式"""
    OVERWRITE = "w"
    APPEND = "a"


class TransactionState(Enum):
    """事务状态"""
    PENDING = auto()
    COMMITTED = auto()
    ROLLED_BACK = auto()


# ============== 数据类定义 ==============

@dataclass
class ReadOptions:
    """读取选项"""
    mode: ReadMode = ReadMode.TEXT
    encoding: str = "utf-8"
    offset: Optional[int] = None
    size: Optional[int] = None
    chunk_size: int = 8192


@dataclass
class WriteOptions:
    """写入选项"""
    mode: WriteMode = WriteMode.OVERWRITE
    encoding: str = "utf-8"
    create_dirs: bool = True
    backup: bool = True
    atomic: bool = True
    fsync: bool = True


@dataclass
class ReadResult:
    """读取结果"""
    content: Union[str, bytes]
    size: int
    checksum: str
    encoding: Optional[str] = None


@dataclass
class WriteResult:
    """写入结果"""
    path: Path
    size: int
    checksum: str
    backup_path: Optional[Path] = None


@dataclass
class OperationRecord:
    """操作记录"""
    op_type: str
    path: Path
    undo_data: Any
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class Transaction:
    """事务"""
    id: str
    operations: list[OperationRecord] = field(default_factory=list)
    state: TransactionState = TransactionState.PENDING
    committed_at: Optional[datetime] = None


@dataclass
class LockInfo:
    """锁信息"""
    owner: str
    acquired_at: float
    expires_at: Optional[float] = None


# ============== 异常定义 ==============

class FileAgentError(Exception):
    """File Agent 基础异常"""
    pass


class FileReadError(FileAgentError):
    """文件读取错误"""
    pass


class FileWriteError(FileAgentError):
    """文件写入错误"""
    pass


class SecurityError(FileAgentError):
    """安全错误"""
    pass


class TransactionError(FileAgentError):
    """事务错误"""
    pass


# ============== 核心 File Agent 类 ==============

class FileAgent:
    """
    完整的 File Agent 实现
    
    Features:
    - 文件操作原语（Read/Write/Create/Delete/Rename）
    - 事务性批量操作与回滚
    - 撤销/重做支持
    - 文件锁与乐观锁
    - 语义感知（文件类型、编码检测）
    - 安全边界防护
    """
    
    def __init__(
        self,
        base_path: Union[str, Path],
        enable_undo: bool = True,
        max_undo_stack: int = 100,
        max_file_size: int = 100 * 1024 * 1024
    ):
        """
        初始化 File Agent
        
        Args:
            base_path: 允许访问的基础路径
            enable_undo: 是否启用撤销/重做
            max_undo_stack: 最大撤销栈深度
            max_file_size: 最大文件大小（字节）
        """
        self.base_path = Path(base_path).resolve()
        self.enable_undo = enable_undo
        self.max_undo_stack = max_undo_stack
        self.max_file_size = max_file_size
        
        # 内部状态
        self._undo_stack: list[OperationRecord] = []
        self._redo_stack: list[OperationRecord] = []
        self._transaction_stack: list[Transaction] = []
        self._locks: dict[str, LockInfo] = {}
        
        # 确保 base_path 存在
        if not self.base_path.exists():
            self.base_path.mkdir(parents=True, exist_ok=True)
    
    # ============== 读取操作 ==============
    
    async def read(
        self,
        path: Union[str, Path],
        options: Optional[ReadOptions] = None
    ) -> ReadResult:
        """
        读取文件
        
        Args:
            path: 文件路径
            options: 读取选项
            
        Returns:
            ReadResult: 读取结果
        """
        options = options or ReadOptions()
        resolved = self._resolve_path(path)
        
        self._validate_path(resolved)
        
        if not await aiofiles.os.path.exists(resolved):
            raise FileReadError(f"File not found: {resolved}")
        
        # 检查文件大小
        stat = await aiofiles.os.stat(resolved)
        if stat.st_size > self.max_file_size:
            raise FileReadError(f"File too large: {stat.st_size} bytes")
        
        # 执行读取
        mode = "r" if options.mode == ReadMode.TEXT else "rb"
        enc = options.encoding if options.mode == ReadMode.TEXT else None
        
        async with aiofiles.open(resolved, mode=mode, encoding=enc) as f:
            if options.offset is not None:
                await f.seek(options.offset)
            
            if options.size is not None:
                content = await f.read(options.size)
            else:
                content = await f.read()
        
        # 计算校验和
        data = content.encode() if isinstance(content, str) else content
        checksum = hashlib.sha256(data).hexdigest()
        
        return ReadResult(
            content=content,
            size=len(data),
            checksum=checksum,
            encoding=enc
        )
    
    async def read_lines(
        self,
        path: Union[str, Path],
        line_start: Optional[int] = None,
        line_end: Optional[int] = None
    ) -> AsyncIterator[str]:
        """
        按行读取
        
        Args:
            path: 文件路径
            line_start: 起始行号（从0开始）
            line_end: 结束行号
            
        Yields:
            str: 文件行内容
        """
        result = await self.read(path)
        content = result.content if isinstance(result.content, str) else result.content.decode()
        lines = content.split('\n')
        
        start = line_start or 0
        end = line_end or len(lines)
        
        for i in range(start, min(end, len(lines))):
            yield lines[i]
    
    # ============== 写入操作 ==============
    
    async def write(
        self,
        path: Union[str, Path],
        content: Union[str, bytes],
        options: Optional[WriteOptions] = None
    ) -> WriteResult:
        """
        写入文件
        
        Args:
            path: 文件路径
            content: 文件内容
            options: 写入选项
            
        Returns:
            WriteResult: 写入结果
        """
        options = options or WriteOptions()
        resolved = self._resolve_path(path)
        
        self._validate_path(resolved)
        
        # 计算数据
        data = content.encode() if isinstance(content, str) else content
        checksum = hashlib.sha256(data).hexdigest()
        
        backup_path = None
        
        # 确保目录存在
        if options.create_dirs:
            await self._ensure_parent_dir(resolved)
        
        # 备份现有文件
        if options.backup and await aiofiles.os.path.exists(resolved):
            backup_path = await self._create_backup(resolved)
        
        # 写入
        if options.atomic:
            temp_path = await self._write_atomic(resolved, data, options)
        else:
            temp_path = await self._write_direct(resolved, data, options)
        
        # 记录撤销信息
        if self.enable_undo:
            self._record_operation(
                OperationRecord(
                    op_type='write',
                    path=resolved,
                    undo_data=backup_path.read_bytes() if backup_path and backup_path.exists() else None
                )
            )
        
        return WriteResult(
            path=temp_path,
            size=len(data),
            checksum=checksum,
            backup_path=backup_path
        )
    
    async def _write_atomic(
        self,
        path: Path,
        data: bytes,
        options: WriteOptions
    ) -> Path:
        """原子性写入"""
        fd, temp_path_str = tempfile.mkstemp(
            dir=path.parent,
            prefix=f".{path.name}.",
            suffix=".tmp"
        )
        temp_path = Path(temp_path_str)
        
        try:
            loop = asyncio.get_event_loop()
            await loop.run_in_executor(None, lambda: os.write(fd, data))
            
            if options.fsync:
                await loop.run_in_executor(None, lambda: os.fsync(fd))
            
            await loop.run_in_executor(None, lambda: os.close(fd))
            
            # 原子性重命名
            await asyncio.to_thread(
                lambda: shutil.move(str(temp_path), str(path))
            )
            
            return path
            
        except Exception:
            try:
                await asyncio.to_thread(lambda: os.close(fd))
            except:
                pass
            if temp_path.exists():
                temp_path.unlink()
            raise
    
    async def _write_direct(
        self,
        path: Path,
        data: bytes,
        options: WriteOptions
    ) -> Path:
        """直接写入"""
        mode = "wb" if isinstance(data, bytes) else "w"
        enc = None if isinstance(data, bytes) else options.encoding
        
        async with aiofiles.open(path, mode=mode, encoding=enc) as f:
            await f.write(data if isinstance(data, str) else data.decode())
            
            if options.fsync:
                await f.flush()
                await asyncio.to_thread(lambda: os.fsync(f.fileno()))
        
        return path
    
    # ============== 文件操作 ==============
    
    async def create(
        self,
        path: Union[str, Path],
        is_dir: bool = False,
        parents: bool = True,
        mode: Optional[int] = None
    ) -> Path:
        """
        创建文件或目录
        
        Args:
            path: 创建路径
            is_dir: 是否为目录
            parents: 是否创建父目录
            mode: 权限模式
            
        Returns:
            Path: 创建的路径
        """
        resolved = self._resolve_path(path)
        self._validate_path(resolved)
        
        if is_dir:
            await asyncio.to_thread(
                lambda: resolved.mkdir(parents=parents, exist_ok=True, mode=mode or 0o755)
            )
        else:
            if parents:
                await self._ensure_parent_dir(resolved)
            await asyncio.to_thread(lambda: resolved.touch(mode=mode))
        
        if self.enable_undo:
            self._record_operation(
                OperationRecord(op_type='create', path=resolved, undo_data=None)
            )
        
        return resolved
    
    async def delete(
        self,
        path: Union[str, Path],
        recursive: bool = False,
        backup: bool = True
    ) -> None:
        """
        删除文件或目录
        
        Args:
            path: 删除路径
            recursive: 是否递归删除
            backup: 是否备份
        """
        resolved = self._resolve_path(path)
        self._validate_path(resolved)
        
        if not await aiofiles.os.path.exists(resolved):
            return
        
        # 保存删除数据用于撤销
        undo_data = None
        if self.enable_undo and backup:
            stat = await aiofiles.os.stat(resolved)
            is_dir = stat.st_mode & 0o170000 == 0o40000
            if is_dir:
                undo_data = {'type': 'dir', 'path': str(resolved)}
            else:
                async with aiofiles.open(resolved, 'rb') as f:
                    undo_data = {'type': 'file', 'data': await f.read()}
        
        # 执行删除
        stat = await aiofiles.os.stat(resolved)
        is_dir = stat.st_mode & 0o170000 == 0o40000
        
        if is_dir:
            if recursive:
                await asyncio.to_thread(lambda: shutil.rmtree(resolved))
            else:
                await asyncio.to_thread(lambda: resolved.rmdir())
        else:
            await asyncio.to_thread(lambda: resolved.unlink())
        
        if self.enable_undo:
            self._record_operation(
                OperationRecord(op_type='delete', path=resolved, undo_data=undo_data)
            )
    
    async def rename(
        self,
        src: Union[str, Path],
        dst: Union[str, Path],
        overwrite: bool = False
    ) -> Path:
        """
        重命名或移动
        
        Args:
            src: 源路径
            dst: 目标路径
            overwrite: 是否覆盖目标
            
        Returns:
            Path: 新路径
        """
        src_resolved = self._resolve_path(src)
        dst_resolved = self._resolve_path(dst)
        
        self._validate_path(src_resolved)
        self._validate_path(dst_resolved)
        
        if not await aiofiles.os.path.exists(src_resolved):
            raise FileWriteError(f"Source not found: {src_resolved}")
        
        if await aiofiles.os.path.exists(dst_resolved) and not overwrite:
            raise FileWriteError(f"Destination exists: {dst_resolved}")
        
        await asyncio.to_thread(
            lambda: shutil.move(str(src_resolved), str(dst_resolved))
        )
        
        if self.enable_undo:
            self._record_operation(
                OperationRecord(
                    op_type='rename',
                    path=dst_resolved,
                    undo_data={'src': str(src_resolved), 'dst': str(dst_resolved)}
                )
            )
        
        return dst_resolved
    
    # ============== 撤销/重做 ==============
    
    async def undo(self) -> Optional[str]:
        """
        撤销上一个操作
        
        Returns:
            Optional[str]: 被撤销的操作类型
        """
        if not self._undo_stack:
            return None
        
        op = self._undo_stack.pop()
        self._redo_stack.append(op)
        
        await self._execute_undo(op)
        return op.op_type
    
    async def redo(self) -> Optional[str]:
        """
        重做上一个被撤销的操作
        
        Returns:
            Optional[str]: 被重做的操作类型
        """
        if not self._redo_stack:
            return None
        
        op = self._redo_stack.pop()
        self._undo_stack.append(op)
        
        await self._execute_redo(op)
        return op.op_type
    
    async def _execute_undo(self, op: OperationRecord) -> None:
        """执行撤销"""
        if op.op_type == 'write':
            if op.undo_data and isinstance(op.undo_data, bytes):
                await self.write(op.path, op.undo_data, WriteOptions(atomic=False))
        elif op.op_type == 'delete':
            if op.undo_data:
                if op.undo_data.get('type') == 'dir':
                    await self.create(op.path, is_dir=True)
                else:
                    await self.write(op.path, op.undo_data['data'], WriteOptions())
        elif op.op_type == 'rename':
            info = op.undo_data
            await self.rename(info['dst'], info['src'])
        elif op.op_type == 'create':
            await self.delete(op.path)
    
    async def _execute_redo(self, op: OperationRecord) -> None:
        """执行重做"""
        if op.op_type == 'write':
            # 需要原始内容才能重做
            pass
        elif op.op_type == 'delete':
            await self.delete(op.path)
        elif op.op_type == 'rename':
            info = op.undo_data
            await self.rename(info['src'], info['dst'])
        elif op.op_type == 'create':
            await self.create(op.path)
    
    def _record_operation(self, op: OperationRecord) -> None:
        """记录操作"""
        self._undo_stack.append(op)
        self._redo_stack.clear()
        
        if len(self._undo_stack) > self.max_undo_stack:
            self._undo_stack.pop(0)
    
    # ============== 事务支持 ==============
    
    @asynccontextmanager
    async def transaction(self):
        """
        事务上下文管理器
        
        Usage:
            async with agent.transaction():
                await agent.write("file1.txt", "content1")
                await agent.write("file2.txt", "content2")
        """
        tx = Transaction(id=str(uuid.uuid4()))
        self._transaction_stack.append(tx)
        
        try:
            yield tx
            await self._commit_transaction(tx)
        except Exception as e:
            await self._rollback_transaction(tx)
            raise TransactionError(f"Transaction failed: {e}") from e
        finally:
            if tx in self._transaction_stack:
                self._transaction_stack.remove(tx)
    
    async def _commit_transaction(self, tx: Transaction) -> None:
        """提交事务"""
        tx.committed_at = datetime.now()
        tx.state = TransactionState.COMMITTED
    
    async def _rollback_transaction(self, tx: Transaction) -> None:
        """回滚事务"""
        tx.state = TransactionState.ROLLED_BACK
        # 逆序撤销操作
        for op in reversed(tx.operations):
            await self._execute_undo(op)
    
    # ============== 辅助方法 ==============
    
    def _resolve_path(self, path: Union[str, Path]) -> Path:
        """解析路径"""
        if isinstance(path, str):
            path = Path(path)
        if not path.is_absolute():
            path = self.base_path / path
        return path.resolve()
    
    def _validate_path(self, path: Path) -> None:
        """验证路径安全性"""
        path_str = str(path)
        base_str = str(self.base_path)
        
        if not path_str.startswith(base_str):
            raise SecurityError(f"Path outside base: {path}")
        
        # 检查敏感文件
        sensitive = {'.env', '.ssh', '.git', 'id_rsa', 'id_ed25519', 'authorized_keys'}
        for part in path.parts:
            if part in sensitive:
                raise SecurityError(f"Access to sensitive path denied: {path}")
    
    async def _ensure_parent_dir(self, path: Path) -> None:
        """确保父目录存在"""
        parent = path.parent
        if not await aiofiles.os.path.exists(parent):
            await asyncio.to_thread(lambda: parent.mkdir(parents=True, exist_ok=True))
    
    async def _create_backup(self, path: Path) -> Path:
        """创建备份"""
        backup_dir = path.parent / ".backups"
        await asyncio.to_thread(lambda: backup_dir.mkdir(parents=True, exist_ok=True))
        
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        backup_path = backup_dir / f"{path.name}.{timestamp}.bak"
        
        await asyncio.to_thread(lambda: shutil.copy2(path, backup_path))
        return backup_path


# ============== 使用示例 ==============

async def main():
    """File Agent 使用示例"""
    
    # 创建 File Agent
    agent = FileAgent("./workspace", enable_undo=True)
    
    # 基本读写
    result = await agent.write("test.txt", "Hello, File Agent!")
    print(f"Written: {result.path}, checksum: {result.checksum}")
    
    content = await agent.read("test.txt")
    print(f"Read: {content.content}, size: {content.size}")
    
    # 批量操作
    files = {
        "file1.txt": "Content 1",
        "file2.txt": "Content 2",
        "file3.txt": "Content 3"
    }
    
    async with agent.transaction():
        for path, content in files.items():
            await agent.write(path, content)
        print("Batch write completed")
    
    # 撤销操作
    await agent.undo()
    print("Undo completed")
    
    await agent.redo()
    print("Redo completed")


if __name__ == "__main__":
    asyncio.run(main())