"""SpecAgent - Specification and versioning agent for VoIdea. This agent: - Manages specifications - Handles versioning - Generates CHANGELOG - Updates project.json """ import json import re 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 SpecAgent(BaseAgent): """Specification and versioning agent.""" name = "spec_agent" version = "1.0.0" description = "Manages specifications, versioning, and CHANGELOG" triggers = [ AgentTrigger.MANUAL, AgentTrigger.TAG_CREATION, AgentTrigger.PUSH, ] def __init__(self): super().__init__() self.project_root = Path(__file__).parent.parent.parent self.changelog_dir = self.project_root / "CHANGELOG" self.project_json = self.project_root / "project.json" async def run(self, context: dict[str, Any] | None = None) -> AgentResult: """Execute specification task. Context can contain: - action: str (version_bump, changelog, update_json, check_version) - version_type: str (major, minor, patch) - commit_message: str """ await self.set_running("specification") try: action = context.get("action", "check") if context else "check" if action == "version_bump": result = await self._bump_version(context or {}) elif action == "changelog": result = await self._generate_changelog(context or {}) elif action == "update_json": result = await self._update_project_json(context or {}) elif action == "check": result = await self._check_version() else: result = await self._check_version() await self.set_idle() return result except Exception as e: await self.set_error(str(e)) return AgentResult( success=False, message=f"SpecAgent failed: {str(e)}", errors=[str(e)], ) async def health_check(self) -> bool: """Check if SpecAgent is operational.""" return self.changelog_dir.exists() or self.changelog_dir.mkdir(exist_ok=True) async def _check_version(self) -> AgentResult: """Check current version and changelog status.""" current_version = settings.project_version changelog_files = list(self.changelog_dir.glob("v*.md")) changelog_files.sort(reverse=True) return AgentResult( success=True, message=f"Current version: {current_version}", data={ "current_version": current_version, "changelog_files": [f.name for f in changelog_files], "changelog_count": len(changelog_files), }, ) async def _bump_version(self, context: dict[str, Any]) -> AgentResult: """Bump version based on conventional commits.""" version_type = context.get("version_type", "patch") commit_message = context.get("commit_message", "") current = settings.project_version major, minor, patch = map(int, current.split(".")) if version_type == "major": major += 1 minor = 0 patch = 0 elif version_type == "minor": minor += 1 patch = 0 else: patch += 1 new_version = f"{major}.{minor}.{patch}" changelog_file = self.changelog_dir / f"v{major}.{minor}.md" if not changelog_file.exists(): changelog_file.write_text( f"# Changelog v{major}.{minor}\n\n" f"Generated: {datetime.now(timezone.utc).isoformat()}\n\n" f"## [{new_version}] - {datetime.now().strftime('%Y-%m-%d')}\n\n", encoding="utf-8", ) else: content = changelog_file.read_text(encoding="utf-8") entry = f"\n## [{new_version}] - {datetime.now().strftime('%Y-%m-%d')}\n\n" if commit_message: entry += f"### Changes\n- {commit_message}\n" content += entry changelog_file.write_text(content, encoding="utf-8") return AgentResult( success=True, message=f"Version bumped: {current} -> {new_version}", data={ "old_version": current, "new_version": new_version, "version_type": version_type, "changelog_file": str(changelog_file.name), }, ) async def _generate_changelog(self, context: dict[str, Any]) -> AgentResult: """Generate changelog from commits.""" commits = context.get("commits", []) version = context.get("version", settings.project_version) changelog_content = f"""## [{version}] - {datetime.now().strftime('%Y-%m-%d')} ### Added """ for commit in commits: commit_type = self._parse_commit_type(commit) message = self._parse_commit_message(commit) if commit_type == "feat": changelog_content += f"- Added: {message}\n" elif commit_type == "fix": changelog_content += f"- Fixed: {message}\n" elif commit_type == "docs": changelog_content += f"- Docs: {message}\n" else: changelog_content += f"- {message}\n" return AgentResult( success=True, message=f"Generated changelog for {version}", data={ "version": version, "content": changelog_content, "commits_count": len(commits), }, ) async def _update_project_json(self, context: dict[str, Any]) -> AgentResult: """Update project.json with current state.""" if not self.project_json.exists(): return AgentResult( success=False, message="project.json not found", ) try: data = json.loads(self.project_json.read_text(encoding="utf-8")) if context: for key, value in context.items(): if key in data: data[key] = value data["updated"] = datetime.now(timezone.utc).isoformat() self.project_json.write_text(json.dumps(data, indent=2), encoding="utf-8") return AgentResult( success=True, message="project.json updated", data={"updated_fields": list(context.keys()) if context else []}, ) except Exception as e: return AgentResult( success=False, message=f"Failed to update project.json: {str(e)}", errors=[str(e)], ) def _parse_commit_type(self, commit: str) -> str: """Parse commit type from conventional commit message.""" match = re.match(r"^(\w+)(?:\(.+\))?:", commit) return match.group(1) if match else "other" def _parse_commit_message(self, commit: str) -> str: """Parse commit message without prefix.""" match = re.match(r"^(\w+)(?:\(.+\))?:\s*(.+)", commit) return match.group(2) if match else commit async def get_metrics(self) -> dict[str, Any]: """Get SpecAgent metrics.""" changelog_files = list(self.changelog_dir.glob("v*.md")) if self.changelog_dir.exists() else [] return { "agent_id": self.name, "version": self.version, "status": self.status.value, "changelog_versions": len(changelog_files), }