251 lines
8.0 KiB
Python
251 lines
8.0 KiB
Python
"""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,
|
|
} |