"""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 = ".", max_entries: int = 200) -> str: """List files and directories at the given path (relative to repo root or absolute). Args: path: Directory to list (relative to repo root or absolute). max_entries: Maximum number of entries to return (default 200). Pass a subdirectory path to narrow results when a directory is very large. """ resolved: str = self._resolve_path(path) try: items: list[str] = sorted(os.listdir(resolved)) total: int = len(items) shown: list[str] = items[:max_entries] result: str = "\n".join(shown) if total > max_entries: result += ( f"\n\n[{total} entries total — showing first {max_entries}. " "Pass a subdirectory path to narrow results.]" ) return result 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 _truncate_output( self, text: str, max_chars: int, offset: int, label: str = "Output", ) -> str: """Slice [offset : offset+max_chars] from text and append a paging footer if truncated.""" total: int = len(text) chunk: str = text[offset : offset + max_chars] if offset > 0 or (offset + max_chars) < total: next_offset: int = offset + len(chunk) chunk += ( f"\n\n[{label} truncated — {total} chars total. " f"Showing chars {offset}–{next_offset}. " f"Re-run with output_offset={next_offset} to read more.]" ) return chunk def run_command( self, command: str, timeout: int = 120, max_chars: int = 8000, output_offset: int = 0, ) -> str: """Execute a shell command in the repository workspace and return stdout/stderr. Args: command: Shell command to run. timeout: Seconds before the command is killed (default 120). max_chars: Maximum characters of combined output to return (default 8000). Tail (most recent lines) is preferred because errors appear there. output_offset: Character offset into the full output to start reading from (default 0). Increment by max_chars to page through large output. """ 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 " f"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() raw: str = ( f"Command timed out after {timeout} seconds.\n" f"--- STDOUT ---\n{stdout}\n--- STDERR ---\n{stderr}" ) return self._truncate_output(raw, max_chars, output_offset, "Output") output: str = "" if stdout: output += f"--- STDOUT ---\n{stdout}" if stderr: output += f"\n--- STDERR ---\n{stderr}" if not output: return "Command executed successfully (no output)." prefix: str = ( f"Command failed with exit code {process.returncode}:\n" if process.returncode != 0 else "" ) full: str = prefix + output return self._truncate_output(full, max_chars, output_offset, "Output") except Exception as e: return f"Error running command: {str(e)}" def grep_search( self, pattern: str, path: str = ".", max_lines: int = 100, offset: int = 0, ) -> str: """Search for pattern in files under path using grep (case-insensitive). Args: pattern: Regular-expression / literal pattern to search for. path: Directory or file to search (relative to repo root or absolute). max_lines: Maximum number of matching lines to return (default 100). offset: Line offset into the full result set for paging (default 0). """ 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.\n" f"Stdout: {stdout}\nStderr: {stderr}" ) if process.returncode != 0 and not stdout: return f"No matches found for '{pattern}'." all_lines: list[str] = stdout.splitlines() total: int = len(all_lines) page: list[str] = all_lines[offset : offset + max_lines] result: str = "\n".join(page) if total > offset + max_lines: next_offset: int = offset + max_lines result += ( f"\n\n[{total} matches total — showing lines {offset}–{offset + len(page)}. " f"Use offset={next_offset} to see more.]" ) if stderr: result += f"\nError: {stderr}" return result except Exception as e: return f"Error during grep search: {str(e)}"