""" Aider coding agent tool — invokes Aider AI pair programming as a subprocess. Aider handles repo-map generation, file editing, git commits, and linting automatically. It works with any OpenAI-compatible model — point it at DeepSeek, Ollama, OpenRouter, etc. via AIDER_MODEL / AIDER_OPENAI_API_BASE env vars or the project's .aider.conf.yml. background=True runs the subprocess asynchronously and returns an agent_id immediately. The caller can poll via agent_status() or request a push notification via notify=True. """ import asyncio import logging import os from pathlib import Path from google.genai import types import agent_manager logger = logging.getLogger(__name__) _CORTEX_DIR = Path(__file__).parent # .../Cortex_and_Inara_dev/cortex/ _PROJECT_ROOT = _CORTEX_DIR.parent # .../Cortex_and_Inara_dev/ # Known project aliases — expand before passing to subprocess _PROJECT_ALIASES: dict[str, str] = { "cortex": str(_PROJECT_ROOT), "aether_api": "~/OSIT_dev/aether_api_fastapi", "aether_frontend": "~/OSIT_dev/aether_app_sveltekit", "aether_container": "~/OSIT_dev/aether_container_env", } _MAX_OUTPUT_CHARS = 12_000 async def aider_run( project: str, task: str, files: list[str] | None = None, model: str | None = None, auto_commit: bool = True, timeout: int = 300, background: bool = False, notify: bool = False, ) -> str: """Run Aider with a single task in a project directory, then exit. When background=True, fires the subprocess asynchronously and returns an agent_id immediately. Use agent_status(agent_id) to check progress; set notify=True to receive a push/Talk notification on completion. """ resolved = _PROJECT_ALIASES.get(project, project) cwd = Path(os.path.expanduser(resolved)) if not cwd.is_dir(): return f"Error: project directory '{resolved}' does not exist." timeout = min(max(int(timeout), 10), 600) cmd: list[str] = [ "aider", "--message", task, "--yes-always", "--no-pretty", "--no-stream", "--no-check-update", "--no-detect-urls", "--auto-commits" if auto_commit else "--no-auto-commits", ] # Inject OpenRouter credentials from the Cortex model registry if available. # Aider's subprocess inherits Cortex's environment, which doesn't include keys # stored in ~/.env or shell profiles. Pulling from the registry keeps it self-contained. try: import model_registry from persona import get_user user = get_user() or "scott" registry = model_registry.get_registry(user) or_host = next( (h for h in registry.get("hosts", []) if "openrouter.ai" in h.get("api_url", "")), None, ) if or_host and or_host.get("api_key"): cmd += ["--api-key", f"openrouter={or_host['api_key']}"] except Exception: user = "scott" # non-fatal — user may have key via env or .aider.conf.yml if model: cmd += ["--model", model] for f in (files or []): cmd += ["--file", f] logger.info( "aider_run: project=%s model=%s auto_commit=%s files=%s background=%s task=%.120s", project, model, auto_commit, files, background, task, ) async def _run() -> str: proc = await asyncio.create_subprocess_exec( *cmd, cwd=str(cwd), stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=float(timeout)) out = stdout.decode(errors="replace").strip() err = stderr.decode(errors="replace").strip() parts = [] if out: parts.append(out) if err: parts.append(f"[stderr]\n{err}") combined = "\n".join(parts) if parts else "(no output)" if len(combined) > _MAX_OUTPUT_CHARS: half = _MAX_OUTPUT_CHARS // 2 combined = ( combined[:half] + f"\n\n[... {len(combined) - _MAX_OUTPUT_CHARS} chars trimmed ...]\n\n" + combined[-half:] ) if proc.returncode not in (0, 1): return f"[exit {proc.returncode}]\n{combined}" return combined if background: rec = await agent_manager.register( user=user, role="aider", task=task, level=2, notify=notify, ) async def _bg_task() -> None: try: result = await _run() await agent_manager.finish(rec.agent_id, result, "done") logger.info("aider_run [bg]: done %s", rec.agent_id[:8]) except asyncio.CancelledError: await agent_manager.finish(rec.agent_id, "Cancelled.", "cancelled") raise except asyncio.TimeoutError: msg = f"Aider timed out after {timeout}s" logger.warning("aider_run [bg]: timeout %s", rec.agent_id[:8]) await agent_manager.finish(rec.agent_id, msg, "timeout") except FileNotFoundError: msg = "Error: 'aider' not found in PATH — run: pip install aider-chat" await agent_manager.finish(rec.agent_id, msg, "failed") except Exception as e: logger.error("aider_run [bg]: failed %s: %s", rec.agent_id[:8], e) await agent_manager.finish(rec.agent_id, str(e), "failed") bg = asyncio.create_task(_bg_task()) agent_manager.set_task_ref(rec.agent_id, bg) return ( f"Aider task started in background. ID: {rec.agent_id}\n" f"Use agent_status('{rec.agent_id}') to monitor progress." ) # Synchronous path try: return await _run() except asyncio.TimeoutError: return f"Error: aider timed out after {timeout}s" except FileNotFoundError: return "Error: 'aider' not found in PATH — run: pip install aider-chat" except Exception as e: logger.error("aider_run error: %s", e) return f"Error: {e}" DECLARATIONS = [ types.FunctionDeclaration( name="aider_run", description=( "Run the Aider AI coding agent on a project with a single task, then exit. " "Aider maps the repo, edits files, runs lint checks, and optionally commits. " "Use for code changes, bug fixes, refactoring, or new features across any " "configured project. Model is set via AIDER_MODEL env var or .aider.conf.yml " "in the project directory — no API key needed if the project is already configured. " "Set background=True for long tasks — returns an agent_id immediately and sends " "a notification when done. ADMIN ONLY. Requires confirmation." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "project": types.Schema( type=types.Type.STRING, description=( "Project alias or absolute path. Known aliases: " "'cortex' (this project), 'aether_api', 'aether_frontend', " "'aether_container'. Or provide an absolute path like " "'/home/scott/OSIT_dev/aether_api_fastapi'." ), ), "task": types.Schema( type=types.Type.STRING, description=( "Full task description sent to Aider as --message. " "Be specific — include file names, what to change, and why. " "Example: 'In cortex/tools/web.py, add a max_chars parameter " "to web_read() capped at 32768.'" ), ), "files": types.Schema( type=types.Type.ARRAY, items=types.Schema(type=types.Type.STRING), description=( "Optional list of files to add explicitly to the editing context " "(paths relative to the project root). " "Aider also builds a repo map automatically — these get priority." ), ), "model": types.Schema( type=types.Type.STRING, description=( "Optional model override. Examples: 'deepseek/deepseek-chat', " "'openrouter/anthropic/claude-3-5-haiku-20241022'. " "Defaults to the project's .aider.conf.yml model or AIDER_MODEL env var." ), ), "auto_commit": types.Schema( type=types.Type.BOOLEAN, description=( "Auto-commit changes after edits (default: true). " "Set to false to review diffs before committing manually." ), ), "timeout": types.Schema( type=types.Type.INTEGER, description="Max seconds to wait for Aider to finish (default 300, max 600).", ), "background": types.Schema( type=types.Type.BOOLEAN, description=( "Run asynchronously in the background (default: false). " "Returns an agent_id immediately; use agent_status(agent_id) to monitor. " "Recommended for tasks expected to take more than ~60 seconds." ), ), "notify": types.Schema( type=types.Type.BOOLEAN, description=( "Send a push/Talk notification when the background task completes " "(default: false). Only applies when background=true." ), ), }, required=["project", "task"], ), ) ]