Spaces:
Sleeping
Sleeping
File size: 3,319 Bytes
8b02e7c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | """Aider CLI adapter."""
import asyncio
import subprocess
from typing import Any
from .base import CLIAdapter
class AiderAdapter(CLIAdapter):
"""Adapter for Aider CLI."""
async def execute(self, task: str, progress_callback: Any = None, timeout: int | None = None, **kwargs: Any) -> tuple[str, str, int]:
"""Execute task using Aider CLI with optional streaming."""
if progress_callback:
return await self.execute_streaming(task, progress_callback, timeout, **kwargs)
cmd = self.format_task(task, **kwargs)
resolved_cmd = self.resolve_command(cmd)
# Sanitize environment for Windows/prompt_toolkit compatibility
env = {**subprocess.os.environ, **self.get_env()}
env["TERM"] = "dumb" # Prevent prompt_toolkit from seeing xterm-256color on Windows
env["PYTHONIOENCODING"] = "utf-8"
process = await asyncio.create_subprocess_exec(
*resolved_cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
env=env,
)
try:
effective_timeout = timeout or self.get_timeout()
stdout, stderr = await asyncio.wait_for(
process.communicate(), timeout=effective_timeout
)
return (
stdout.decode("utf-8", errors="replace"),
stderr.decode("utf-8", errors="replace"),
process.returncode or 0,
)
except asyncio.TimeoutError:
process.kill()
await process.wait()
raise TimeoutError(f"Aider CLI timed out after {effective_timeout}s")
def get_env(self) -> dict[str, str]:
"""Get environment variables for Aider."""
env = super().get_env().copy()
env["TERM"] = "dumb" # Prevent prompt_toolkit from seeing xterm-256color on Windows
env["PYTHONIOENCODING"] = "utf-8"
return env
def validate(self) -> bool:
"""Validate Aider is available."""
try:
subprocess.run(
["which", "aider"] if subprocess.os.name != "nt" else ["where", "aider"],
capture_output=True,
check=True,
)
return True
except subprocess.CalledProcessError:
return False
def format_task(self, task: str, **kwargs: Any) -> list[str]:
"""Format task for Aider CLI."""
cmd = ["aider"]
# Add message flag for non-interactive mode
cmd.append("--message")
cmd.append(task)
# Add model if specified
if model := kwargs.get("model"):
cmd.extend(["--model", model])
# Add mode flags
if kwargs.get("architect_mode"):
cmd.append("--architect")
elif kwargs.get("ask_mode"):
cmd.append("--ask")
# Auto-commit changes
if kwargs.get("auto_commit", True):
cmd.append("--auto-commits")
# Add optimization flags
cmd.extend([
"--no-pretty",
"--stream",
"--no-check-update",
"--no-show-release-notes",
"--verbose",
"--yes-always",
])
# Add any custom args from config
cmd.extend(self.get_args())
return cmd
|