"""Agent orchestrator for software generation.""" import asyncio from typing import Optional from ai_software_factory.agents.git_manager import GitManager from ai_software_factory.agents.ui_manager import UIManager from ai_software_factory.agents.gitea import GiteaAPI from ai_software_factory.agents.database_manager import DatabaseManager from ai_software_factory.config import settings from datetime import datetime import os class AgentOrchestrator: """Orchestrates the software generation process with full audit trail.""" def __init__( self, project_id: str, project_name: str, description: str, features: list, tech_stack: list, db = None ): """Initialize orchestrator.""" self.project_id = project_id self.project_name = project_name self.description = description self.features = features self.tech_stack = tech_stack self.status = "initialized" self.progress = 0 self.current_step = None self.message = "" self.logs = [] self.ui_data = {} self.db = db # Initialize agents self.git_manager = GitManager(project_id) self.ui_manager = UIManager(project_id) self.gitea_api = GiteaAPI( token=settings.GITEA_TOKEN, base_url=settings.GITEA_URL, owner=settings.GITEA_OWNER, repo=settings.GITEA_REPO or "" ) # Initialize database manager if db session provided self.db_manager = None self.history = None if db: self.db_manager = DatabaseManager(db) # Log project start to database self.history = self.db_manager.log_project_start( project_id=project_id, project_name=project_name, description=description ) # Re-fetch with new history_id self.db_manager = DatabaseManager(db) async def run(self) -> dict: """Run the software generation process with full audit logging.""" try: # Step 1: Initialize project self.progress = 5 self.current_step = "Initializing project" self.message = "Setting up project structure..." self.logs.append(f"[{datetime.utcnow().isoformat()}] Initializing project.") # Step 2: Create project structure (skip git operations) self.progress = 15 self.current_step = "Creating project structure" self.message = "Creating project files..." await self._create_project_structure() # Step 3: Generate initial code self.progress = 25 self.current_step = "Generating initial code" self.message = "Generating initial code with Ollama..." await self._generate_code() # Step 4: Test the code self.progress = 50 self.current_step = "Testing code" self.message = "Running tests..." await self._run_tests() # Step 5: Commit to git (skip in test env) self.progress = 75 self.current_step = "Committing to git" self.message = "Skipping git operations in test environment..." # Step 6: Create PR (skip in test env) self.progress = 90 self.current_step = "Creating PR" self.message = "Skipping PR creation in test environment..." # Step 7: Complete self.progress = 100 self.current_step = "Completed" self.message = "Software generation complete!" self.logs.append(f"[{datetime.utcnow().isoformat()}] Software generation complete!") # Log completion to database if available if self.db_manager and self.history: self.db_manager.log_project_complete( history_id=self.history.id, message="Software generation complete!" ) return { "status": "completed", "progress": self.progress, "message": self.message, "current_step": self.current_step, "logs": self.logs, "ui_data": self.ui_manager.ui_data, "history_id": self.history.id if self.history else None } except Exception as e: self.status = "error" self.message = f"Error: {str(e)}" self.logs.append(f"[{datetime.utcnow().isoformat()}] Error: {str(e)}") # Log error to database if available if self.db_manager and self.history: self.db_manager.log_error( history_id=self.history.id, error=str(e) ) return { "status": "error", "progress": self.progress, "message": self.message, "current_step": self.current_step, "logs": self.logs, "error": str(e), "ui_data": self.ui_manager.ui_data, "history_id": self.history.id if self.history else None } async def _create_project_structure(self) -> None: """Create initial project structure.""" project_dir = self.project_id # Create .gitignore gitignore_path = f"{project_dir}/.gitignore" try: os.makedirs(project_dir, exist_ok=True) with open(gitignore_path, "w") as f: f.write("# Python\n__pycache__/\n*.pyc\n*.pyo\n*.pyd\n.Python\n*.env\n.venv/\nnode_modules/\n.env\nbuild/\ndist/\n.pytest_cache/\n.mypy_cache/\n.coverage\nhtmlcov/\n.idea/\n.vscode/\n*.swp\n*.swo\n*~\n.DS_Store\n.git\n") except Exception as e: self.logs.append(f"[{datetime.utcnow().isoformat()}] Failed to create .gitignore: {str(e)}") # Create README.md readme_path = f"{project_dir}/README.md" try: with open(readme_path, "w") as f: f.write(f"# {self.project_name}\n\n{self.description}\n\n## Features\n") for feature in self.features: f.write(f"- {feature}\n") f.write(f"\n## Tech Stack\n") for tech in self.tech_stack: f.write(f"- {tech}\n") except Exception as e: self.logs.append(f"[{datetime.utcnow().isoformat()}] Failed to create README.md: {str(e)}") async def _generate_code(self) -> None: """Generate code using Ollama.""" # This would call Ollama API to generate code # For now, create a placeholder file try: main_py_path = f"{self.project_id}/main.py" os.makedirs(self.project_id, exist_ok=True) with open(main_py_path, "w") as f: f.write("# Generated by AI Software Factory\n") f.write("print('Hello, World!')\n") except Exception as e: self.logs.append(f"[{datetime.utcnow().isoformat()}] Failed to create main.py: {str(e)}") # Log code change to audit trail if self.db_manager and self.history: self.db_manager.log_code_change( project_id=self.project_id, change_type="CREATE", file_path="main.py", actor="agent", actor_type="agent", details="Generated main.py file" ) async def _run_tests(self) -> None: """Run tests for the generated code.""" # This would run pytest or other test framework # For now, simulate test success pass async def _commit_to_git(self) -> None: """Commit changes to git.""" pass # Skip git operations in test environment async def _create_pr(self) -> None: """Create pull request.""" pass # Skip PR creation in test environment def update_status(self, status: str, progress: int, message: str) -> None: """Update status and progress.""" self.status = status self.progress = progress self.message = message def get_ui_data(self) -> dict: """Get UI data.""" return self.ui_manager.ui_data def render_dashboard(self) -> str: """Render dashboard HTML.""" return self.ui_manager.render_dashboard() def get_history(self) -> Optional[dict]: """Get project history from database.""" if self.db_manager and self.history: return self.db_manager.get_project_audit_data(self.history.project_id) return None