hendrik/tools/coder.py

import os
import subprocess
import re
import fnmatch
import glob as glob_module

# Global state to track the agent's current working directory
# This ensures that subsequent bash commands run in the correct directory
CURRENT_WORKSPACE = os.getcwd()

def get_current_workspace():
    global CURRENT_WORKSPACE
    return CURRENT_WORKSPACE

def set_current_workspace(path):
    global CURRENT_WORKSPACE
    CURRENT_WORKSPACE = path
    os.chdir(path)

schema_read_file = {
    "type": "function",
    "function": {
        "name"         : "read_file",
        "description"  : "Read the contents of a file. Returns the file content with line numbers prefixed.",
        "parameters"   : {
            "type"        : "object",
            "properties"  : {
                "path": {"type": "string", "description": "Absolute path to the file to read"}
            },
            "required"    : ["path"]
        }
    }
}

def read_file(path):
    try:
        # Use absolute path if it's relative to CURRENT_WORKSPACE
        full_path = os.path.join(get_current_workspace(), path) if not os.path.isabs(path) else path
        with open(full_path, 'r', encoding='utf-8', errors='replace') as f:
            lines = f.readlines()
        return ''.join(f"{i+1}: {line}" for i, line in enumerate(lines))
    except Exception as e:
        return f"Error reading file: {str(e)}"

schema_write_file = {
    "type": "function",
    "function": {
        "name"         : "write_file",
        "description"  : "Write content to a file. Overwrites the file if it exists.",
        "parameters"   : {
            "type"        : "object",
            "properties"  : {
                "path": {"type": "string", "description": "Absolute path to the file to write"},
                "content": {"type": "string", "description": "Content to write to the file"}
            },
            "required"    : ["path", "content"]
        }
    }
}

def write_file(path, content):
    try:
        full_path = os.path.join(get_current_workspace(), path) if not os.path.isabs(path) else path
        with open(full_path, 'w', encoding='utf-8') as f:
            f.write(content)
        return f"Successfully wrote to {full_path}"
    except Exception as e:
        return f"Error writing file: {str(e)}"

schema_edit_file = {
    "type": "function",
    "function": {
        "name"         : "edit_file",
        "description"  : "Replace old_string with new_string in a file. If old_string is not found, returns error. If multiple matches, replaces all.",
        "parameters"   : {
            "type"        : "object",
            "properties"  : {
                "path": {"type": "string", "description": "Absolute path to the file to edit"},
                "old_string": {"type": "string", "description": "String to replace"},
                "new_string": {"type": "string", "description": "Replacement string"}
            },
            "required"    : ["path", "old_string", "new_string"]
        }
    }
}

def edit_file(path, old_string, new_string):
    try:
        full_path = os.path.join(get_current_workspace(), path) if not os.path.isabs(path) else path
        with open(full_path, 'r', encoding='utf-8', errors='replace') as f:
            content = f.read()
        if old_string not in content:
            return f"Error: old_string not found in {full_path}"
        new_content = content.replace(old_string, new_string)
        with open(full_path, 'w', encoding='utf-8') as f:
            f.write(new_content)
        return f"Successfully edited {full_path}"
    except Exception as e:
        return f"Error editing file: {str(e)}"

schema_run_bash = {
    "type": "function",
    "function": {
        "name"         : "run_bash",
        "description"  : "Run a bash command. Returns stdout, stderr, and return code. Commands are executed relative to the current workspace.",
        "parameters"   : {
            "type"        : "object",
            "properties"  : {
                "command": {"type": "string", "description": "Bash command to execute"}
            },
            "required"    : ["command"]
        }
    }
}

def run_bash(command):
    try:
        # Prepend 'cd' to the command to ensure it runs in the correct workspace
        full_command = f"cd {get_current_workspace()} && {command}"
        result = subprocess.run(full_command, shell=True, capture_output=True, text=True, timeout=30)
        return f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}\nreturn code: {result.returncode}"
    except Exception as e:
        return f"Error running command: {str(e)}"

schema_search_code = {
    "type": "function",
    "function": {
        "name": "search_code",
        "description": "Search for files or content in files. Use type 'glob' to find files matching a pattern, 'content' to search file contents for a regex pattern.",
        "parameters": {
            "type": "object",
            "properties": {
                "pattern": {"type": "string", "description": "Glob pattern (for type 'glob') or regex pattern (for type 'content')"},
                "path": {"type": "string", "description": "Directory to search in (default: current working directory)", "default": "."},
                "search_type": {"type": "string", "enum": ["glob", "content"], "description": "Type of search: 'glob' for files, 'content' for file contents"}
            },
            "required": ["pattern", "search_type"]
        }
    }
}

def search_code(pattern, search_type, path="."):
    try:
        # Resolve path relative to workspace
        search_path = os.path.join(get_current_workspace(), path) if not os.path.isabs(path) else path
        
        if search_type == "glob":
            files = glob_module.glob(f"{search_path}/**/{pattern}", recursive=True)
            return "\n".join(files) if files else "No files found"
        elif search_type == "content":
            results = []
            for root, dirs, files in os.walk(search_path):
                for file in files:
                    file_path = os.path.join(root, file)
                    try:
                        with open(file_path, 'r', encoding='utf-8', errors='replace') as f:
                            for i, line in enumerate(f.readlines(), 1):
                                if re.search(pattern, line):
                                    results.append(f"{file_path}:{i}: {line.strip()}")
                    except:
                        continue
            return "\n".join(results[:50]) if results else "No matches found"
        else:
            return "Invalid search_type. Use 'glob' or 'content'."
    except Exception as e:
        return f"Error searching: {str(e)}"
    
schema_git_operation = {
    "type": "function",
    "function": {
        "name": "git_operation",
        "description": "Run a git command. Pass the git arguments as a list (e.g., ['status', '--short'] for 'git status --short'). "
                       "POLICY: Never run 'git add' or 'git commit' without explicit user permission. "
                       "Safe to run without asking: git status, git diff, git log. "
                       "Always ask first before committing.",
        "parameters": {
            "type": "object",
            "properties": {
                "args": {"type": "array", "items": {"type": "string"}, "description": "List of git command arguments (without 'git' prefix)"}
            },
            "required": ["args"]
        }
    }
}

def git_operation(args):
    try:
        # Execute git command from the current workspace
        full_command = ["git", *args]
        result = subprocess.run(full_command, cwd=get_current_workspace(), capture_output=True, text=True, timeout=10)
        return f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}\nreturn code: {result.returncode}"
    except Exception as e:
        return f"Error running git command: {str(e)}"

schema_workspace_change = {
    "type": "function",
    "function": {
        "name": "workspace_change",
        "description": "Change the current working directory of the agent session. Supports tilde (~) expansion for the user home directory (/home/aji/).",
        "parameters": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "The target directory path to change to."}
            },
            "required": ["path"]
        }
    }
}

def workspace_change(path):
    try:
        if path.startswith("~"):
            target_path = os.path.expanduser(path)
        else:
            target_path = os.path.abspath(path)

        if os.path.exists(target_path) and os.path.isdir(target_path):
            set_current_workspace(target_path)
            return f"Successfully changed workspace to: {target_path}"
        else:
            return f"Error: The path {target_path} does not exist or is not a directory."
    except Exception as e:
        return f"Error changing workspace: {str(e)}"

schema_search_grep = {
    "type": "function",
    "function": {
        "name": "search_grep",
        "description": "Search file contents using regular expressions. Fast content search across your codebase. Supports full regex syntax and file pattern filtering.",
        "parameters": {
            "type": "object",
            "properties": {
                "pattern": {"type": "string", "description": "Regex pattern to search for in file contents"},
                "path": {"type": "string", "description": "Directory to search in (default: current working directory)", "default": "."},
                "include": {"type": "string", "description": "File pattern to include (e.g. '*.py', '*.{ts,tsx}'). If not provided, all files are searched."}
            },
            "required": ["pattern"]
        }
    }
}

def search_grep(pattern, path=".", include=None):
    try:
        search_path = os.path.join(get_current_workspace(), path) if not os.path.isabs(path) else path
        results = []
        for root, dirs, files in os.walk(search_path):
            for file in files:
                if include and not fnmatch.fnmatch(file, include):
                    continue
                file_path = os.path.join(root, file)
                try:
                    with open(file_path, 'r', encoding='utf-8', errors='replace') as f:
                        for i, line in enumerate(f.readlines(), 1):
                            if re.search(pattern, line):
                                results.append(f"{file_path}:{i}: {line.strip()}")
                except:
                    continue
        return "\n".join(results[:100]) if results else "No matches found"
    except Exception as e:
        return f"Error searching: {str(e)}"

schema_search_glob = {
    "type": "function",
    "function": {
        "name": "search_glob",
        "description": "Find files by pattern matching. Search for files using glob patterns like `**/*.html` or `modules/**/*.py`. Returns matching file paths sorted by modification time.",
        "parameters": {
            "type": "object",
            "properties": {
                "pattern": {"type": "string", "description": "Glob pattern to match files (e.g. '**/*.html', 'modules/**/*.py')"},
                "path": {"type": "string", "description": "Directory to search in (default: current working directory)", "default": "."}
            },
            "required": ["pattern"]
        }
    }
}

def search_glob(pattern, path="."):
    try:
        search_path = os.path.join(get_current_workspace(), path) if not os.path.isabs(path) else path
        files = glob_module.glob(f"{search_path}/{pattern}", recursive=True)
        def _isfile(f):
            try:
                return os.path.isfile(f)
            except OSError:
                return False
        files = [f for f in files if _isfile(f)]
        def _mtime(f):
            try:
                return os.path.getmtime(f)
            except OSError:
                return 0
        files.sort(key=_mtime, reverse=True)
        return "\n".join(files[:500]) if files else "No files found"
    except Exception as e:
        return f"Error searching: {str(e)}"