91 lines
3.0 KiB
Python
91 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import pwd
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
from .config import Settings
|
|
|
|
|
|
class AiderAgent:
|
|
def __init__(self, settings: Settings) -> None:
|
|
self.settings = settings
|
|
|
|
def run(self, workspace: Path, task: str) -> str:
|
|
task_file = workspace / ".git" / "failure-task.txt"
|
|
task_file.write_text(task)
|
|
task_file.chmod(0o600)
|
|
command = self.command(task_file)
|
|
coder = pwd.getpwnam("coder")
|
|
git_config = Path(coder.pw_dir) / ".gitconfig"
|
|
git_config.write_text(
|
|
"[user]\n\tname = LAI Automation Coder\n\temail = coder@localhost\n"
|
|
f"[safe]\n\tdirectory = {workspace.resolve()}\n"
|
|
)
|
|
git_config.chmod(0o600)
|
|
os.chown(git_config, coder.pw_uid, coder.pw_gid)
|
|
os.chown(task_file, coder.pw_uid, coder.pw_gid)
|
|
for path in workspace.rglob("*"):
|
|
if ".git" not in path.parts:
|
|
os.chown(path, coder.pw_uid, coder.pw_gid, follow_symlinks=False)
|
|
os.chown(workspace, coder.pw_uid, coder.pw_gid, follow_symlinks=False)
|
|
|
|
environment = {
|
|
"HOME": coder.pw_dir,
|
|
"PATH": os.environ.get("PATH", "/usr/local/bin:/usr/bin:/bin"),
|
|
}
|
|
for name in ("MISTRAL_API_KEY", "OPENAI_API_KEY", "ANTHROPIC_API_KEY"):
|
|
if os.getenv(name):
|
|
environment[name] = os.environ[name]
|
|
|
|
def drop_privileges() -> None:
|
|
os.setgroups([])
|
|
os.setgid(coder.pw_gid)
|
|
os.setuid(coder.pw_uid)
|
|
|
|
try:
|
|
result = subprocess.run(
|
|
command,
|
|
cwd=workspace,
|
|
env=environment,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
preexec_fn=drop_privileges,
|
|
timeout=900,
|
|
)
|
|
finally:
|
|
shutil.rmtree(workspace / ".aider.tags.cache.v4", ignore_errors=True)
|
|
for artifact in (".aider.chat.history.md", ".aider.input.history"):
|
|
(workspace / artifact).unlink(missing_ok=True)
|
|
os.chown(workspace, 0, coder.pw_gid, follow_symlinks=False)
|
|
workspace.chmod(0o775)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(f"Aider failed ({result.returncode}): {result.stderr[-4000:]}")
|
|
return result.stdout[-12_000:]
|
|
|
|
def command(self, task_file: Path) -> list[str]:
|
|
return [
|
|
self.settings.agent_command,
|
|
"--model",
|
|
self.settings.agent_model,
|
|
"--message-file",
|
|
str(task_file),
|
|
"--yes",
|
|
"--no-auto-commits",
|
|
"--no-dirty-commits",
|
|
"--no-stream",
|
|
"--no-gitignore",
|
|
"--chat-history-file",
|
|
"/dev/null",
|
|
"--input-history-file",
|
|
"/dev/null",
|
|
"--no-analytics",
|
|
"--edit-format",
|
|
"diff",
|
|
"--timeout",
|
|
"60",
|
|
]
|