Michael Ingvarsson
318 lines
12 KiB
Python
318 lines
12 KiB
Python
"""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 _get_subprocess_env(self) -> dict[str, str]:
|
||
from gitea.config import GITEA_URL, GITEA_TOKEN
|
||
env = os.environ.copy()
|
||
if GITEA_URL:
|
||
env["GITEA_SERVER_URL"] = GITEA_URL
|
||
if GITEA_TOKEN:
|
||
env["GITEA_SERVER_TOKEN"] = GITEA_TOKEN
|
||
return env
|
||
|
||
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,
|
||
env=self._get_subprocess_env()
|
||
)
|
||
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,
|
||
env=self._get_subprocess_env(),
|
||
)
|
||
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: list[str] = ["grep", "-ri", pattern, resolved]
|
||
process: subprocess.Popen[str] = subprocess.Popen(
|
||
command,
|
||
shell=False,
|
||
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)}"
|