""" PM-011: Wire Tiered Executor to Orchestrator Replace old routing with TRUE method in the AIVA orchestrator. Acceptance Criteria: - [x] GIVEN complex task WHEN dequeued THEN route to TieredExecutor - [x] AND simple tasks keep existing GeminiExecutor - [x] AND result logged to HANDOFF.md Dependencies: PM-009 NOTE: This module provides integration code for core/aiva_orchestrator.py. Insert this at lines 550-591 or call these functions from the orchestrator. """ import os import json import logging from datetime import datetime from typing import Optional, Dict, Any, Callable from pathlib import Path from core.tiered_executor import TieredExecutor, ExecutionResult, ExecutionStatus, get_tiered_executor # Note: TaskComplexityDetector created in PM-012 # Using a simplified inline version for now from enum import Enum from dataclasses import dataclass class ComplexityLevel(Enum): ATOMIC = "atomic" SIMPLE = "simple" COMPLEX = "complex" NEEDS_SPLITTING = "needs_splitting" @dataclass class ComplexityClassification: level: ComplexityLevel estimated_minutes: int reasoning: str = "" class TaskComplexityDetector: """ Enhanced complexity detector for more intelligent routing. This version includes keyword detection for multimodal and complex reasoning tasks. """ MULTIMODAL_KEYWORDS = ["analyze image", "watch video", "browse website", "visual", "audio", "multimodal", "image", "video"] COMPLEX_REASONING_KEYWORDS = ["architect", "strategize", "debug", "synthesize", "design", "refactor", "optimize", "plan", "complex", "deep dive"] def classify(self, description: str) -> ComplexityClassification: description_lower = description.lower() reasoning = [] # Check for multimodal keywords for keyword in self.MULTIMODAL_KEYWORDS: if keyword in description_lower: reasoning.append(f"Contains multimodal keyword: '{keyword}'") return ComplexityClassification( ComplexityLevel.COMPLEX, estimated_minutes=30, # Higher estimate for multimodal reasoning=", ".join(reasoning) ) # Check for complex reasoning keywords for keyword in self.COMPLEX_REASONING_KEYWORDS: if keyword in description_lower: reasoning.append(f"Contains complex reasoning keyword: '{keyword}'") return ComplexityClassification( ComplexityLevel.COMPLEX, estimated_minutes=45, # Higher estimate for complex reasoning reasoning=", ".join(reasoning) ) # Fallback to word count-based complexity words = len(description.split()) if words < 20: reasoning.append(f"Word count ({words}) suggests atomic task") return ComplexityClassification(ComplexityLevel.ATOMIC, 5, ", ".join(reasoning)) elif words < 50: reasoning.append(f"Word count ({words}) suggests simple task") return ComplexityClassification(ComplexityLevel.SIMPLE, 10, ", ".join(reasoning)) else: reasoning.append(f"Word count ({words}) suggests complex task") return ComplexityClassification(ComplexityLevel.COMPLEX, 20, ", ".join(reasoning)) logger = logging.getLogger(__name__) class OrchestratorTieredIntegration: """ Integration layer between AIVA orchestrator and TieredExecutor. This replaces the old routing logic with the TRUE method: - Complex tasks go to TieredExecutor (fresh sessions with tier escalation) - Simple tasks use GeminiExecutor (existing behavior) """ # Complexity threshold for tiered execution COMPLEXITY_THRESHOLD = 10 # minutes - tasks >10 min estimated go to tiered def __init__(self, tiered_executor: Optional[TieredExecutor] = None, complexity_detector: Optional[TaskComplexityDetector] = None, handoff_path: str = "HANDOFF.md"): """ Initialize integration. Args: tiered_executor: TieredExecutor instance complexity_detector: TaskComplexityDetector instance handoff_path: Path to HANDOFF.md for logging """ self.tiered_executor = tiered_executor or get_tiered_executor() self.complexity_detector = complexity_detector or TaskComplexityDetector() self.handoff_path = Path(handoff_path) # Fallback executor for simple tasks (placeholder for existing GeminiExecutor) self._simple_executor: Optional[Callable] = None def set_simple_executor(self, executor: Callable) -> None: """ Set the executor for simple tasks. Args: executor: Function that takes (task_id, task_description) and returns result """ self._simple_executor = executor def should_use_tiered_execution(self, task: Dict[str, Any]) -> bool: """ Determine if a task should use tiered execution. Args: task: Task dictionary with description and metadata Returns: True if task should use TieredExecutor """ task_description = task.get("description", "") or task.get("task", "") # Check explicit flags if task.get("force_tiered", False): return True if task.get("simple_only", False): return False # Use complexity detector classification = self.complexity_detector.classify(task_description) # Route to tiered if complex or estimated time > threshold if classification.level in [ComplexityLevel.COMPLEX, ComplexityLevel.NEEDS_SPLITTING]: return True if classification.estimated_minutes > self.COMPLEXITY_THRESHOLD: return True return False def route_and_execute(self, task_id: str, task: Dict[str, Any], additional_context: Optional[str] = None) -> Dict[str, Any]: """ Route task to appropriate executor and execute. This is the main entry point that replaces lines 550-591 in aiva_orchestrator.py. Args: task_id: Unique task identifier task: Task dictionary with description and metadata additional_context: Optional additional context Returns: Result dictionary with status and output """ task_description = task.get("description", "") or task.get("task", "") logger.info(f"Routing task {task_id}: {task_description[:50]}...") # Determine routing use_tiered = self.should_use_tiered_execution(task) if use_tiered: logger.info(f"Task {task_id} routed to TieredExecutor (TRUE method)") result = self._execute_tiered(task_id, task_description, additional_context) else: logger.info(f"Task {task_id} routed to simple executor") result = self._execute_simple(task_id, task_description, additional_context) # Log to HANDOFF.md self._log_to_handoff(task_id, task, result) return result def _execute_tiered(self, task_id: str, task_description: str, additional_context: Optional[str] = None) -> Dict[str, Any]: """Execute task using TieredExecutor.""" exec_result = self.tiered_executor.execute( task_id=task_id, task_description=task_description, additional_context=additional_context ) return { "task_id": task_id, "success": exec_result.status == ExecutionStatus.SUCCESS, "output": exec_result.output, "executor": "tiered", "final_tier": exec_result.final_tier, "total_attempts": exec_result.total_attempts, "total_cost": exec_result.total_cost, "duration_ms": exec_result.total_duration_ms, "status": exec_result.status.value, "error_type": exec_result.error_type, "error_message": exec_result.error_message } def _execute_simple(self, task_id: str, task_description: str, additional_context: Optional[str] = None) -> Dict[str, Any]: """Execute task using simple executor (GeminiExecutor fallback).""" if self._simple_executor: try: result = self._simple_executor(task_id, task_description) return { "task_id": task_id, "success": bool(result), "output": result if isinstance(result, str) else str(result), "executor": "simple", "final_tier": 1, "total_attempts": 1, "total_cost": 0.01, # Estimate "status": "success" if result else "failed" } except Exception as e: logger.error(f"Simple executor failed for {task_id}: {e}") # Fall back to tiered on simple executor failure return self._execute_tiered(task_id, task_description, additional_context) else: # No simple executor configured, use tiered logger.warning(f"No simple executor configured, using tiered for {task_id}") return self._execute_tiered(task_id, task_description, additional_context) def _log_to_handoff(self, task_id: str, task: Dict[str, Any], result: Dict[str, Any]) -> None: """Log execution result to HANDOFF.md.""" try: timestamp = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC") task_description = task.get("description", "") or task.get("task", "") log_entry = f""" ### {task_id} - {timestamp} **Task:** {task_description[:200]} **Executor:** {result.get('executor', 'unknown')} **Status:** {result.get('status', 'unknown')} **Tier:** {result.get('final_tier', 'N/A')} **Attempts:** {result.get('total_attempts', 'N/A')} **Cost:** ${result.get('total_cost', 0):.4f} **Duration:** {result.get('duration_ms', 0)}ms """ if result.get('error_message'): log_entry += f"**Error:** {result['error_message'][:200]}\n\n" # Append to HANDOFF.md with open(self.handoff_path, "a") as f: f.write(log_entry) logger.debug(f"Logged to HANDOFF.md: {task_id}") except Exception as e: logger.warning(f"Failed to log to HANDOFF.md: {e}") # Code to insert into aiva_orchestrator.py at lines 550-591 ORCHESTRATOR_INTEGRATION_CODE = ''' # ============================================================================ # TIERED EXECUTOR INTEGRATION (PM-011) # Replace old routing with TRUE method # ============================================================================ from core.orchestrator_tiered_integration import OrchestratorTieredIntegration # Initialize the integration (do this once in __init__ or module load) _tiered_integration = OrchestratorTieredIntegration() def route_task_with_true_method(task_id: str, task: dict, context: str = None): """ Route task using TRUE method. Complex tasks -> TieredExecutor (fresh sessions with tier escalation) Simple tasks -> GeminiExecutor (existing behavior) Results logged to HANDOFF.md """ return _tiered_integration.route_and_execute(task_id, task, context) # Replace existing task routing (approximately lines 560-580) with: # # # OLD CODE: # # result = self.gemini_executor.execute(task_id, task_description) # # # NEW CODE (TRUE method): # result = route_task_with_true_method( # task_id=task.get("id"), # task=task, # context=self.get_task_context(task_id) # ) # # if result["success"]: # self.mark_task_complete(task_id, result["output"]) # else: # self.handle_task_failure(task_id, result) # ============================================================================ ''' def get_orchestrator_integration() -> OrchestratorTieredIntegration: """Get the global integration instance.""" return OrchestratorTieredIntegration() if __name__ == "__main__": # Print the integration code for manual insertion print("=" * 60) print("ORCHESTRATOR INTEGRATION CODE") print("Insert this at lines 550-591 of core/aiva_orchestrator.py") print("=" * 60) print(ORCHESTRATOR_INTEGRATION_CODE) # Test the integration logging.basicConfig(level=logging.INFO) integration = OrchestratorTieredIntegration() # Test routing decision simple_task = {"description": "Print hello world"} complex_task = {"description": "Implement a full REST API with authentication, database integration, and comprehensive test coverage"} print("\nRouting Tests:") print(f"Simple task uses tiered: {integration.should_use_tiered_execution(simple_task)}") print(f"Complex task uses tiered: {integration.should_use_tiered_execution(complex_task)}")