hkuds--deeptutor/deeptutor/tools/file_tools.py

"""Workspace file tools for chat skill/script workflows."""

from __future__ import annotations

import difflib
from pathlib import Path
from typing import Any

from deeptutor.core.tool_protocol import BaseTool, ToolDefinition, ToolParameter, ToolResult
from deeptutor.tools.prompting import load_prompt_hints


def _resolve_workspace_path(path: str, workspace: str, allowed_dir: str) -> Path:
    if not workspace or not allowed_dir:
        raise PermissionError("workspace file tools are not available for this turn")
    raw = Path(str(path or ".")).expanduser()
    root = Path(workspace).expanduser().resolve()
    allowed = Path(allowed_dir).expanduser().resolve()
    candidate = raw if raw.is_absolute() else root / raw
    resolved = candidate.resolve()
    try:
        resolved.relative_to(allowed)
    except ValueError as exc:
        raise PermissionError(f"path is outside the turn workspace: {path}") from exc
    return resolved


class _WorkspaceTool(BaseTool):
    def get_prompt_hints(self, language: str = "en"):
        return load_prompt_hints(self.name, language=language)

    @staticmethod
    def _resolve(kwargs: dict[str, Any], path: str) -> Path:
        return _resolve_workspace_path(
            path,
            str(kwargs.get("_workspace_dir") or ""),
            str(kwargs.get("_allowed_dir") or ""),
        )


class ReadFileTool(_WorkspaceTool):
    _DEFAULT_LIMIT = 2000
    _MAX_CHARS = 128_000

    def get_definition(self) -> ToolDefinition:
        return ToolDefinition(
            name="read_file",
            description="Read a text file from this turn's workspace with line pagination.",
            parameters=[
                ToolParameter(name="path", type="string", description="Path inside the workspace."),
                ToolParameter(
                    name="offset",
                    type="integer",
                    description="1-indexed line number to start from.",
                    required=False,
                ),
                ToolParameter(
                    name="limit",
                    type="integer",
                    description="Maximum lines to return.",
                    required=False,
                ),
            ],
        )

    async def execute(self, **kwargs: Any) -> ToolResult:
        path = str(kwargs.get("path") or "")
        try:
            fp = self._resolve(kwargs, path)
            if not fp.exists():
                return ToolResult(content=f"Error: file not found: {path}", success=False)
            if not fp.is_file():
                return ToolResult(content=f"Error: not a file: {path}", success=False)
            try:
                offset = max(1, int(kwargs.get("offset") or 1))
            except (TypeError, ValueError):
                offset = 1
            try:
                limit = max(1, int(kwargs.get("limit") or self._DEFAULT_LIMIT))
            except (TypeError, ValueError):
                limit = self._DEFAULT_LIMIT
            lines = fp.read_text(encoding="utf-8", errors="replace").splitlines()
            total = len(lines)
            if total == 0:
                return ToolResult(content=f"(empty file: {path})")
            if offset > total:
                return ToolResult(
                    content=f"Error: offset {offset} is beyond end of file ({total} lines)",
                    success=False,
                )
            start = offset - 1
            end = min(start + limit, total)
            numbered = [f"{start + i + 1}| {line}" for i, line in enumerate(lines[start:end])]
            text = "\n".join(numbered)
            if len(text) > self._MAX_CHARS:
                text = text[: self._MAX_CHARS].rstrip() + "\n...[truncated]"
            suffix = (
                f"\n\n(showing lines {offset}-{end} of {total}; use offset={end + 1} to continue)"
                if end < total
                else f"\n\n(end of file; {total} lines total)"
            )
            return ToolResult(content=text + suffix)
        except PermissionError as exc:
            return ToolResult(content=f"Error: {exc}", success=False)
        except Exception as exc:
            return ToolResult(content=f"Error reading file: {exc}", success=False)


class WriteFileTool(_WorkspaceTool):
    def get_definition(self) -> ToolDefinition:
        return ToolDefinition(
            name="write_file",
            description="Write a UTF-8 text file inside this turn's workspace.",
            parameters=[
                ToolParameter(name="path", type="string", description="Path inside the workspace."),
                ToolParameter(name="content", type="string", description="File contents."),
            ],
        )

    async def execute(self, **kwargs: Any) -> ToolResult:
        path = str(kwargs.get("path") or "")
        content = str(kwargs.get("content") or "")
        try:
            fp = self._resolve(kwargs, path)
            fp.parent.mkdir(parents=True, exist_ok=True)
            fp.write_text(content, encoding="utf-8")
            return ToolResult(content=f"Successfully wrote {len(content)} bytes to {fp.name}")
        except PermissionError as exc:
            return ToolResult(content=f"Error: {exc}", success=False)
        except Exception as exc:
            return ToolResult(content=f"Error writing file: {exc}", success=False)


class EditFileTool(_WorkspaceTool):
    def get_definition(self) -> ToolDefinition:
        return ToolDefinition(
            name="edit_file",
            description="Edit a workspace text file by replacing old_text with new_text.",
            parameters=[
                ToolParameter(name="path", type="string", description="Path inside the workspace."),
                ToolParameter(name="old_text", type="string", description="Text to replace."),
                ToolParameter(name="new_text", type="string", description="Replacement text."),
                ToolParameter(
                    name="replace_all",
                    type="boolean",
                    description="Replace all occurrences instead of one.",
                    required=False,
                ),
            ],
        )

    async def execute(self, **kwargs: Any) -> ToolResult:
        path = str(kwargs.get("path") or "")
        old_text = str(kwargs.get("old_text") or "")
        new_text = str(kwargs.get("new_text") or "")
        replace_all = bool(kwargs.get("replace_all") or False)
        try:
            fp = self._resolve(kwargs, path)
            if not fp.exists():
                return ToolResult(content=f"Error: file not found: {path}", success=False)
            content = fp.read_text(encoding="utf-8", errors="replace")
            if old_text not in content:
                return ToolResult(
                    content=_not_found_message(path, old_text, content),
                    success=False,
                )
            count = content.count(old_text)
            if count > 1 and not replace_all:
                return ToolResult(
                    content=(
                        f"Warning: old_text appears {count} times. Provide more context "
                        "or set replace_all=true."
                    ),
                    success=False,
                )
            updated = (
                content.replace(old_text, new_text)
                if replace_all
                else content.replace(old_text, new_text, 1)
            )
            fp.write_text(updated, encoding="utf-8")
            return ToolResult(content=f"Successfully edited {fp.name}")
        except PermissionError as exc:
            return ToolResult(content=f"Error: {exc}", success=False)
        except Exception as exc:
            return ToolResult(content=f"Error editing file: {exc}", success=False)


class ListDirTool(_WorkspaceTool):
    _DEFAULT_MAX = 200
    _IGNORE = {
        ".git",
        "node_modules",
        "__pycache__",
        ".venv",
        "venv",
        "dist",
        "build",
        ".pytest_cache",
        ".ruff_cache",
    }

    def get_definition(self) -> ToolDefinition:
        return ToolDefinition(
            name="list_dir",
            description="List files under this turn's workspace.",
            parameters=[
                ToolParameter(
                    name="path",
                    type="string",
                    description="Directory path inside the workspace.",
                    required=False,
                    default=".",
                ),
                ToolParameter(
                    name="recursive",
                    type="boolean",
                    description="Recursively list nested paths.",
                    required=False,
                ),
                ToolParameter(
                    name="max_entries",
                    type="integer",
                    description="Maximum entries to return.",
                    required=False,
                ),
            ],
        )

    async def execute(self, **kwargs: Any) -> ToolResult:
        path = str(kwargs.get("path") or ".")
        recursive = bool(kwargs.get("recursive") or False)
        try:
            cap = max(1, int(kwargs.get("max_entries") or self._DEFAULT_MAX))
        except (TypeError, ValueError):
            cap = self._DEFAULT_MAX
        try:
            dp = self._resolve(kwargs, path)
            if not dp.exists():
                return ToolResult(content=f"Error: directory not found: {path}", success=False)
            if not dp.is_dir():
                return ToolResult(content=f"Error: not a directory: {path}", success=False)
            items: list[str] = []
            total = 0
            iterator = dp.rglob("*") if recursive else dp.iterdir()
            for item in sorted(iterator):
                if any(part in self._IGNORE for part in item.relative_to(dp).parts):
                    continue
                total += 1
                if len(items) < cap:
                    rel = item.relative_to(dp)
                    items.append(f"{rel}/" if item.is_dir() else str(rel))
            if not items and total == 0:
                return ToolResult(content=f"Directory {path} is empty")
            text = "\n".join(items)
            if total > cap:
                text += f"\n\n(truncated, showing first {cap} of {total} entries)"
            return ToolResult(content=text)
        except PermissionError as exc:
            return ToolResult(content=f"Error: {exc}", success=False)
        except Exception as exc:
            return ToolResult(content=f"Error listing directory: {exc}", success=False)


def _not_found_message(path: str, old_text: str, content: str) -> str:
    lines = content.splitlines(keepends=True)
    old_lines = old_text.splitlines(keepends=True)
    window = max(1, len(old_lines))
    best_ratio = 0.0
    best_start = 0
    for idx in range(max(1, len(lines) - window + 1)):
        ratio = difflib.SequenceMatcher(None, old_lines, lines[idx : idx + window]).ratio()
        if ratio > best_ratio:
            best_ratio = ratio
            best_start = idx
    if best_ratio > 0.5:
        diff = "\n".join(
            difflib.unified_diff(
                old_lines,
                lines[best_start : best_start + window],
                fromfile="old_text",
                tofile=f"{path} actual line {best_start + 1}",
                lineterm="",
            )
        )
        return f"Error: old_text not found. Best match ({best_ratio:.0%} similar):\n{diff}"
    return "Error: old_text not found. Verify the file content."


__all__ = ["EditFileTool", "ListDirTool", "ReadFileTool", "WriteFileTool"]