"""AuditAgent - Code quality and rules compliance for VoIdea. This agent: - Checks code style (ruff) - Checks type hints (mypy) - Monitors project progress - Verifies documentation compliance """ import asyncio import subprocess from datetime import datetime, timezone from pathlib import Path from typing import Any from app.agents.base import AgentResult, AgentStatus, AgentTrigger, BaseAgent from app.core.config import get_settings settings = get_settings() class AuditAgent(BaseAgent): """Code quality and compliance audit agent.""" name = "audit_agent" version = "1.0.0" description = "Monitors code quality and rule compliance" triggers = [ AgentTrigger.MANUAL, AgentTrigger.PRE_COMMIT, AgentTrigger.CRON, ] def __init__(self): super().__init__() self.project_root = Path(__file__).parent.parent.parent async def run(self, context: dict[str, Any] | None = None) -> AgentResult: """Execute audit task. Context can contain: - action: str (full, quick, style, types, docs) - paths: list[str] (paths to audit) """ await self.set_running("audit") try: action = context.get("action", "full") if context else "full" paths = context.get("paths", ["app"]) if action == "full": result = await self._run_full_audit(paths) elif action == "quick": result = await self._run_quick_audit(paths) elif action == "style": result = await self._run_style_check(paths) elif action == "types": result = await self._run_type_check(paths) elif action == "docs": result = await self._check_docs() else: result = await self._run_quick_audit(paths) await self.set_idle() return result except Exception as e: await self.set_error(str(e)) return AgentResult( success=False, message=f"AuditAgent failed: {str(e)}", errors=[str(e)], ) async def health_check(self) -> bool: """Check if AuditAgent is operational.""" try: return self.project_root.exists() except Exception: return False async def _run_full_audit(self, paths: list[str]) -> AgentResult: """Run full audit including all checks.""" style_result = await self._run_style_check(paths) types_result = await self._run_type_check(paths) docs_result = await self._check_docs() issues = [] issues.extend(style_result.data.get("issues", [])) issues.extend(types_result.data.get("issues", [])) issues.extend(docs_result.data.get("issues", [])) passed = ( style_result.success and types_result.success and docs_result.success ) return AgentResult( success=passed, message=f"Full audit {'passed' if passed else 'failed'}: {len(issues)} issues", data={ "style_check": style_result.data, "type_check": types_result.data, "docs_check": docs_result.data, "total_issues": len(issues), }, ) async def _run_quick_audit(self, paths: list[str]) -> AgentResult: """Run quick audit (ruff only).""" return await self._run_style_check(paths) async def _run_style_check(self, paths: list[str]) -> AgentResult: """Run code style check with ruff.""" issues = [] try: for path in paths: path_obj = self.project_root / path if not path_obj.exists(): issues.append(f"Path not found: {path}") continue result = subprocess.run( ["python", "-m", "ruff", "check", str(path_obj)], capture_output=True, text=True, cwd=str(self.project_root), ) if result.stdout: for line in result.stdout.split("\n"): if line.strip(): issues.append(line.strip()) return AgentResult( success=len(issues) == 0, message=f"Style check: {len(issues)} issues" if issues else "Style check passed", data={ "tool": "ruff", "paths": paths, "issues": issues[:50], "total_issues": len(issues), }, ) except FileNotFoundError: return AgentResult( success=True, message="Ruff not installed, skipping style check", data={"tool": "ruff", "skipped": True}, ) except Exception as e: return AgentResult( success=False, message=f"Style check failed: {str(e)}", errors=[str(e)], ) async def _run_type_check(self, paths: list[str]) -> AgentResult: """Run type checking with mypy.""" issues = [] try: for path in paths: path_obj = self.project_root / path if not path_obj.exists(): continue result = subprocess.run( ["python", "-m", "mypy", str(path_obj), "--ignore-missing-imports"], capture_output=True, text=True, cwd=str(self.project_root), ) if result.stdout: for line in result.stdout.split("\n"): if "error:" in line.lower() or "warning:" in line.lower(): issues.append(line.strip()) return AgentResult( success=len(issues) == 0, message=f"Type check: {len(issues)} issues" if issues else "Type check passed", data={ "tool": "mypy", "paths": paths, "issues": issues[:50], "total_issues": len(issues), }, ) except FileNotFoundError: return AgentResult( success=True, message="Mypy not installed, skipping type check", data={"tool": "mypy", "skipped": True}, ) except Exception as e: return AgentResult( success=False, message=f"Type check failed: {str(e)}", errors=[str(e)], ) async def _check_docs(self) -> AgentResult: """Check documentation completeness.""" missing_docs = [] docs_dir = self.project_root / "docs" app_dir = self.project_root / "app" required_docs = [ "blocks/00-rules.md", "blocks/PLAN.md", "instructions/00-system-prompt.md", ] for doc in required_docs: if not (docs_dir / doc).exists(): missing_docs.append(doc) module_readmes = [ "core/README.md", "models/README.md", "api/README.md", "services/README.md", ] for readme in module_readmes: if not (app_dir / readme).exists(): missing_docs.append(f"app/{readme}") return AgentResult( success=len(missing_docs) == 0, message=f"Documentation check: {len(missing_docs)} missing files", data={ "missing_docs": missing_docs, "total_missing": len(missing_docs), }, ) async def get_metrics(self) -> dict[str, Any]: """Get AuditAgent metrics.""" return { "agent_id": self.name, "version": self.version, "status": self.status.value, "last_run": self.last_run.isoformat() if self.last_run else None, }