coding-agent-gitea/gitea/tools/coding_tools.py

"""Tools for a coding agent to interact with the filesystem and environment."""

import os
import subprocess
from typing import Any


class CodingTools:
    """Tools for a coding agent to interact with the filesystem and environment."""

    def __init__(self, repo_path: str | None = None) -> None:
        self.repo_path: str = repo_path or os.getcwd()

    def get_working_directory(self) -> str:
        """Get the absolute path of the current local repository workspace directory."""
        return self.repo_path

    def _resolve_path(self, path: str) -> str:
        """Resolve a path relative to self.repo_path."""
        if os.path.isabs(path):
            return path
        return os.path.abspath(os.path.join(self.repo_path, path))

    def list_files(self, path: str = ".") -> str:
        """List all files and directories at the given path (relative to repo root or absolute)."""
        resolved: str = self._resolve_path(path)
        try:
            items: list[str] = os.listdir(resolved)
            return "\n".join(items)
        except Exception as e:
            return f"Error listing files: {str(e)}"

    def read_file(self, path: str, offset: int = 1, limit: int = 250) -> str:
        """Read lines from a file, starting at line offset (1-indexed), up to limit lines. Default limit is 250 lines to prevent token bloat. Use 'offset' to scroll/page through larger files."""
        resolved: str = self._resolve_path(path)
        try:
            with open(resolved, 'r', encoding='utf-8', errors='replace') as f:
                lines: list[str] = f.readlines()

            start_line: int = offset - 1
            end_line: int = offset + limit - 1

            content_lines: list[str] = lines[start_line:end_line]

            if not content_lines:
                return "File is empty or offset out of bounds."

            formatted_lines: list[str] = [f"{i + 1}: {line}" for i, line in enumerate(content_lines, start=start_line)]
            return "\n".join(formatted_lines)
        except Exception as e:
            return f"Error reading file: {str(e)}"

    def write_file(self, path: str, content: str) -> str:
        """Write content to a file, creating directories as needed."""
        resolved: str = self._resolve_path(path)
        try:
            os.makedirs(os.path.dirname(os.path.abspath(resolved)), exist_ok=True)
            with open(resolved, 'w', encoding='utf-8') as f:
                f.write(content)
            return f"File {path} written successfully."
        except Exception as e:
            return f"Error writing file: {str(e)}"

    def edit_file(self, path: str, old_content: str, new_content: str) -> str:
        """Replace occurrences of old_content with new_content in the file."""
        resolved: str = self._resolve_path(path)
        try:
            with open(resolved, 'r', encoding='utf-8', errors='replace') as f:
                content: str = f.read()
            if old_content not in content:
                return f"Error: The specified old content was not found in {path}."

            new_content_full: str = content.replace(old_content, new_content)
            with open(resolved, 'w', encoding='utf-8') as f:
                f.write(new_content_full)
            return f"File {path} edited successfully."
        except Exception as e:
            return f"Error editing file: {str(e)}"

    def _parse_verification_commands(self) -> list[str]:
        agents_md: str = os.path.join(self.repo_path, "AGENTS.md")
        if not os.path.exists(agents_md):
            return []
        try:
            with open(agents_md, "r", encoding="utf-8") as f:
                content = f.read()
        except Exception:
            return []

        commands: list[str] = []
        in_verification_section = False
        in_code_block = False
        current_block: list[str] = []

        for line in content.splitlines():
            line_lower = line.strip().lower()
            if line.startswith("#"):
                if "verification" in line_lower or "test" in line_lower:
                    in_verification_section = True
                else:
                    in_verification_section = False
                continue

            if in_verification_section:
                if line.strip().startswith("```"):
                    if in_code_block:
                        in_code_block = False
                        full_cmd = "\n".join(current_block).strip()
                        if full_cmd:
                            for cmd in full_cmd.splitlines():
                                if cmd.strip() and not cmd.strip().startswith("#"):
                                    commands.append(cmd.strip())
                        current_block = []
                    else:
                        in_code_block = True
                elif in_code_block:
                    current_block.append(line)

        return commands

    def run_verification(self) -> tuple[bool, str]:
        commands = self._parse_verification_commands()
        if not commands:
            return True, "No verification commands found in AGENTS.md."

        log_output = []
        for cmd in commands:
            log_output.append(f"Running: {cmd}")
            try:
                res = subprocess.run(
                    cmd, shell=True, cwd=self.repo_path,
                    capture_output=True, text=True, timeout=120
                )
                if res.returncode != 0:
                    log_output.append(
                        f"Command '{cmd}' failed with exit code {res.returncode}:\n"
                        f"Stdout:\n{res.stdout}\n"
                        f"Stderr:\n{res.stderr}"
                    )
                    return False, "\n".join(log_output)
                log_output.append(res.stdout or "Success")
            except subprocess.TimeoutExpired:
                log_output.append(f"Command '{cmd}' timed out after 120 seconds.")
                return False, "\n".join(log_output)
            except Exception as e:
                log_output.append(f"Failed to execute command '{cmd}': {e}")
                return False, "\n".join(log_output)
        return True, "\n".join(log_output)

    def run_command(self, command: str, timeout: int = 120) -> str:
        """Execute a shell command in the repository workspace and return stdout and stderr output. Args: command, timeout (default 120 seconds)."""
        if "tea pr create" in command:
            success, log_msg = self.run_verification()
            if not success:
                return f"Verification failed! You cannot create a pull request because the tests/checks are failing:\n\n{log_msg}\n\nPlease fix the issues and try again."

        try:
            process: subprocess.Popen[str] = subprocess.Popen(
                command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=self.repo_path
            )
            stdout: str
            stderr: str
            try:
                stdout, stderr = process.communicate(timeout=timeout)
            except subprocess.TimeoutExpired:
                process.kill()
                stdout, stderr = process.communicate()
                return f"Command timed out after {timeout} seconds.\n--- STDOUT ---\n{stdout}\n--- STDERR ---\n{stderr}"

            output: str = ""
            if stdout:
                output += f"--- STDOUT ---\n{stdout}"
            if stderr:
                output += f"\n--- STDERR ---\n{stderr}"

            if process.returncode != 0:
                return f"Command failed with exit code {process.returncode}:\n{output}"

            return output if output else "Command executed successfully (no output)."
        except Exception as e:
            return f"Error running command: {str(e)}"

    def grep_search(self, pattern: str, path: str = ".") -> str:
        """Search for pattern in files under path using grep (case-insensitive)."""
        resolved: str = self._resolve_path(path)
        try:
            command: str = f"grep -ri '{pattern}' {resolved}"
            process: subprocess.Popen[str] = subprocess.Popen(
                command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=self.repo_path
            )
            stdout: str
            stderr: str
            try:
                stdout, stderr = process.communicate(timeout=30)
            except subprocess.TimeoutExpired:
                process.kill()
                stdout, stderr = process.communicate()
                return f"Grep search timed out after 30 seconds.\nStdout: {stdout}\nStderr: {stderr}"

            if process.returncode != 0 and not stdout:
                return f"No matches found for '{pattern}'."

            output: str = stdout
            if stderr:
                output += f"\nError: {stderr}"
            return output
        except Exception as e:
            return f"Error during grep search: {str(e)}"