Source code for folderbot.tools.registry

"""Central tool registry for folder tools."""

import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Protocol

import pathspec

from ..bot import BotContext, FolderBot
from .base import ToolResult




class FolderServicesProtocol(Protocol):
    """Protocol for folder tool services/dependencies."""

    @property
    def root(self) -> Path:
        """Root folder path."""
        ...

    @property
    def config(self) -> Any:
        """Configuration object."""
        ...

    @property
    def activity_logger(self) -> Any:
        """Activity logger instance."""
        ...

    @property
    def session_manager(self) -> Any | None:
        """Session manager for user preferences."""
        ...



    def validate_path(self, relative_path: str) -> Path | None:
        """Validate and resolve a path within root folder."""
        ...




    def is_file_allowed(self, rel_path: str) -> bool:
        """Check if file matches include/exclude patterns."""
        ...




    def is_append_allowed(self, rel_path: str) -> bool:
        """Check if file matches append_allowed patterns."""
        ...






@dataclass
class FolderServices:
    """Concrete implementation of folder services."""

    root: Path
    config: Any
    activity_logger: Any
    session_manager: Any | None = None
    _include_spec: Any = field(default=None, repr=False)
    _exclude_spec: Any = field(default=None, repr=False)
    _append_spec: Any = field(default=None, repr=False)

    def __post_init__(self) -> None:
        """Initialize path specs from config."""
        self._include_spec = pathspec.PathSpec.from_lines(
            "gitignore", self.config.read_rules.include
        )
        self._exclude_spec = pathspec.PathSpec.from_lines(
            "gitignore", self.config.read_rules.exclude
        )
        self._append_spec = pathspec.PathSpec.from_lines(
            "gitignore", self.config.read_rules.append_allowed
        )



    def validate_path(self, relative_path: str) -> Path | None:
        """Validate a relative path, ensuring it stays within root folder.

        Returns the logical (un-resolved) path so that symlinks inside the
        root folder work transparently.  Python's pathlib follows symlinks
        automatically for I/O operations (read, write, exists, rglob, etc.).
        Security is enforced by rejecting ``..`` components and absolute paths.
        """
        if not relative_path:
            return self.root

        clean_path = Path(relative_path).as_posix()
        if ".." in clean_path or clean_path.startswith("/"):
            return None

        return self.root / clean_path




    def is_file_allowed(self, rel_path: str) -> bool:
        """Check if file matches include patterns and not exclude patterns."""
        return self._include_spec.match_file(
            rel_path
        ) and not self._exclude_spec.match_file(rel_path)




    def is_append_allowed(self, rel_path: str) -> bool:
        """Check if file matches append_allowed patterns."""
        return self._append_spec.match_file(rel_path)




    def get_tool_config(self, tool_name: str) -> dict[str, Any]:
        """Get configuration for a specific tool from [tools.<name>] section."""
        return self.config.tools.get(tool_name, {})






def get_service(context: BotContext | None, key: str) -> Any:
    """Extract a named service from context."""
    if context is None:
        return None
    return context.services.get(key)





def get_services(context: BotContext | None) -> FolderServices | None:
    """Extract FolderServices from context."""
    return get_service(context, "folder")





def get_tool_config(context: BotContext | None, tool_name: str) -> dict[str, Any]:
    """Get configuration for a specific tool from context."""
    services = get_services(context)
    if services is None:
        return {}
    return services.get_tool_config(tool_name)





def get_root(context: BotContext | None) -> Path | None:
    """Get root folder from context services."""
    services = get_services(context)
    return services.root if services else None





def validate_path(context: BotContext | None, relative_path: str) -> Path | None:
    """Validate path using context services."""
    services = get_services(context)
    if services is None:
        return None
    return services.validate_path(relative_path)





def is_file_allowed(context: BotContext | None, rel_path: str) -> bool:
    """Check if file is allowed using context services."""
    services = get_services(context)
    if services is None:
        return False
    return services.is_file_allowed(rel_path)





def is_append_allowed(context: BotContext | None, rel_path: str) -> bool:
    """Check if append is allowed using context services."""
    services = get_services(context)
    if services is None:
        return False
    return services.is_append_allowed(rel_path)



# Central bot instance for folder tools
folder_bot = FolderBot()




async def execute_with_logging(
    tool_name: str,
    tool_input: dict[str, Any],
    context: BotContext | None = None,
) -> ToolResult:
    """Execute a tool with activity logging."""
    start_time = time.time()
    user_id = context.user_id if context else 0

    try:
        result = await folder_bot.execute_tool(tool_name, tool_input, context)
    except Exception as e:
        result = ToolResult(content=f"Tool execution error: {e}", is_error=True)

    # Log to activity logger if available
    services = get_services(context)
    if services and services.activity_logger:
        duration_ms = int((time.time() - start_time) * 1000)
        # Handle both ToolResult and other response types
        if isinstance(result, ToolResult):
            content = result.content
            is_error = result.is_error
        else:
            content = (
                result.model_dump_json()
                if hasattr(result, "model_dump_json")
                else str(result)
            )
            is_error = False

        services.activity_logger.log_tool_call(
            tool_name=tool_name,
            tool_input=tool_input,
            result=content,
            is_error=is_error,
            user_id=user_id,
            duration_ms=duration_ms,
        )

    return result