Update definitions.py

definitions.py

@@ -25,167 +25,11 @@ import json
 import traceback
 from pathlib import Path
 
-# Set logging level from environment variable
-logging.basicConfig(level=os.getenv('LOG_LEVEL', 'INFO'))
-
-class AutonomousAgentApp:
-    """Main application class for the Autonomous Agent System"""
-
-    def __init__(self):
-        self.workspace_manager = self.WorkspaceManager(workspace_dir=os.getenv('WORKSPACE_DIR', 'workspace'))  # Use self.WorkspaceManager
-        self.pipeline = self._initialize_pipeline()
-        self.refinement_loop = self.RefinementLoop(pipeline=self.pipeline)  # Use self.RefinementLoop
-        self.interface = self.StreamlitInterface(self)  # Use self.StreamlitInterface
-
-    def _initialize_pipeline(self) -> 'AutonomousAgentApp.DevelopmentPipeline':
-        """Initialize the development pipeline"""
-        return self.DevelopmentPipeline(
-            workspace_manager=self.workspace_manager,
-            tool_manager=self._setup_tool_manager()
-        )
-
-    def _setup_tool_manager(self):
-        """Setup tool manager with configuration"""
-        return self.ToolManager()  # Use self.ToolManager
-
-    class ChatSystem:
-        """Manages the chat interaction between users and the autonomous system"""
-        
-        def __init__(self, agent: 'AutonomousAgentApp.AutonomousAgent'):
-            self.agent = agent
-            self.chat_history = []
-            self.active_tasks = {}
-            self.command_handlers = {
-                '/task': self.handle_task_command,
-                '/status': self.handle_status_command,
-                '/stop': self.handle_stop_command,
-                '/help': self.handle_help_command,
-                '/modify': self.handle_modify_command
-            }
-
-        def render_chat_interface(self):
-            """Render the chat interface in Streamlit sidebar"""
-            with st.sidebar:
-                st.markdown("---")
-                st.subheader("System Chat")
-            
-                # Chat controls
-                if st.button("Clear Chat History"):
-                    self.clear_chat_history()
-
-                # Chat history display
-                chat_container = st.container()
-                with chat_container:
-                    for message in self.chat_history:
-                        self._render_message(message)
-                            
-                # Input area
-                user_input = st.text_input("Type message/command...", key="chat_input")
-                if st.button("Send", key="send_message"):
-                    self.process_user_input(user_input)
-                
-    class RefinementLoop:
-        """Manages the iterative refinement process"""
-        
-        def __init__(self, pipeline):
-            self.pipeline = pipeline
-            self.max_iterations = 10
-            self.quality_metrics = QualityMetrics()
-            self.logger = logging.getLogger(__name__)
-            self.current_iteration = 0
-            self.history = []
-
-        async def run_refinement_cycle(self, task: str) -> Dict[str, Any]:
-            """Run a complete refinement cycle for the given task"""
-            self.logger.info(f"Starting refinement cycle for task: {task}")
-            self.current_iteration = 0
-            
-            try:
-                while self.current_iteration < self.max_iterations:
-                    self.logger.info(f"Starting iteration {self.current_iteration + 1}")
-                    
-                    # Execute pipeline stages
-                    planning_result = await self.pipeline.execute_stage(
-                        self.pipeline.PipelineStage.PLANNING,
-                        {"task": task}
-                    )
-                    
-                    development_result = await self.pipeline.execute_stage(
-                        self.pipeline.PipelineStage.DEVELOPMENT,
-                        planning_result["result"]
-                    )
-                    
-                    testing_result = await self.pipeline.execute_stage(
-                        self.pipeline.PipelineStage.TESTING,
-                        development_result["result"]
-                    )
-                    
-                    # Analyze results
-                    quality_analysis = self._analyze_quality(testing_result["result"])
-                    
-                    # Record iteration history
-                    self.history.append({
-                        "iteration": self.current_iteration,
-                        "quality_metrics": quality_analysis,
-                        "timestamp": datetime.now()
-                    })
-                    
-                    # Check if quality requirements are met
-                    if self._meets_quality_requirements(quality_analysis):
-                        self.logger.info("Quality requirements met. Refinement cycle complete.")
-                        return self._prepare_final_result(quality_analysis)
-                    
-                    self.current_iteration += 1
-                
-                return {
-                    "status": "max_iterations_reached",
-                    "iterations_completed": self.current_iteration,
-                    "final_quality": quality_analysis
-                }
-                
-            except Exception as e:
-                self.logger.error(f"Error in refinement cycle: {str(e)}")
-                return {"status": "error", "error": str(e)}
-
-        def _analyze_quality(self, result: Dict[str, Any]) -> Dict[str, float]:
-            """Analyze the quality metrics of the current iteration"""
-            return {
-                "code_quality": self.quality_metrics.code_quality_score,
-                "test_coverage": self.quality_metrics.test_coverage,
-                "security_score": float(self.quality_metrics.security_score)
-            }
 
-        def _meets_quality_requirements(self, quality_analysis: Dict[str, float]) -> bool:
-            """Check if the current quality metrics meet the requirements"""
-            thresholds = self.quality_metrics.thresholds
-            return (
-                quality_analysis["code_quality"] >= thresholds["code_quality"] and
-                quality_analysis["test_coverage"] >= thresholds["test_coverage"] and
-                quality_analysis["security_score"] >= thresholds["security"]
-            )
 
-        def _prepare_final_result(self, quality_analysis: Dict[str, float]) -> Dict[str, Any]:
-            """Prepare the final result of the refinement cycle"""
-            return {
-                "status": "success",
-                "iterations_completed": self.current_iteration,
-                "final_quality": quality_analysis,
-                "history": self.history
-            }
+# Set logging level from environment variable
+logging.basicConfig(level=os.getenv('LOG_LEVEL', 'INFO'))
 
-        def get_refinement_history(self) -> List[Dict[str, Any]]:
-            """Get the history of refinement iterations"""
-            return self.history    
-    
-    def run(self):
-        """Main application entry point"""
-        try:
-            logging.info("Starting Autonomous Agent Application")
-            self.interface.render_main_interface()
-        except Exception as e:
-            logging.error(f"Application error: {str(e)}")
-            st.error("An error occurred while starting the application. Please check the logs.")
-            raise
     class CodeMetricsAnalyzer:
         """Analyzes code metrics using various tools"""
 
@@ -380,6 +224,166 @@ class AutonomousAgentApp:
             "previous": recent_values[0]
         }
 
+class AutonomousAgentApp:
+    """Main application class for the Autonomous Agent System"""
+
+    def __init__(self):
+        self.workspace_manager = self.WorkspaceManager(workspace_dir=os.getenv('WORKSPACE_DIR', 'workspace'))  # Use self.WorkspaceManager
+        self.pipeline = self._initialize_pipeline()
+        self.refinement_loop = self.RefinementLoop(pipeline=self.pipeline)  # Use self.RefinementLoop
+        self.interface = self.StreamlitInterface(self)  # Use self.StreamlitInterface
+
+    def _initialize_pipeline(self) -> 'AutonomousAgentApp.DevelopmentPipeline':
+        """Initialize the development pipeline"""
+        return self.DevelopmentPipeline(
+            workspace_manager=self.workspace_manager,
+            tool_manager=self._setup_tool_manager()
+        )
+
+    def _setup_tool_manager(self):
+        """Setup tool manager with configuration"""
+        return self.ToolManager()  # Use self.ToolManager
+
+    class ChatSystem:
+        """Manages the chat interaction between users and the autonomous system"""
+        
+        def __init__(self, agent: 'AutonomousAgentApp.AutonomousAgent'):
+            self.agent = agent
+            self.chat_history = []
+            self.active_tasks = {}
+            self.command_handlers = {
+                '/task': self.handle_task_command,
+                '/status': self.handle_status_command,
+                '/stop': self.handle_stop_command,
+                '/help': self.handle_help_command,
+                '/modify': self.handle_modify_command
+            }
+
+        def render_chat_interface(self):
+            """Render the chat interface in Streamlit sidebar"""
+            with st.sidebar:
+                st.markdown("---")
+                st.subheader("System Chat")
+            
+                # Chat controls
+                if st.button("Clear Chat History"):
+                    self.clear_chat_history()
+
+                # Chat history display
+                chat_container = st.container()
+                with chat_container:
+                    for message in self.chat_history:
+                        self._render_message(message)
+                            
+                # Input area
+                user_input = st.text_input("Type message/command...", key="chat_input")
+                if st.button("Send", key="send_message"):
+                    self.process_user_input(user_input)
+                
+    class RefinementLoop:
+        """Manages the iterative refinement process"""
+        
+        def __init__(self, pipeline):
+            self.pipeline = pipeline
+            self.max_iterations = 10
+            self.quality_metrics = QualityMetrics()
+            self.logger = logging.getLogger(__name__)
+            self.current_iteration = 0
+            self.history = []
+
+        async def run_refinement_cycle(self, task: str) -> Dict[str, Any]:
+            """Run a complete refinement cycle for the given task"""
+            self.logger.info(f"Starting refinement cycle for task: {task}")
+            self.current_iteration = 0
+            
+            try:
+                while self.current_iteration < self.max_iterations:
+                    self.logger.info(f"Starting iteration {self.current_iteration + 1}")
+                    
+                    # Execute pipeline stages
+                    planning_result = await self.pipeline.execute_stage(
+                        self.pipeline.PipelineStage.PLANNING,
+                        {"task": task}
+                    )
+                    
+                    development_result = await self.pipeline.execute_stage(
+                        self.pipeline.PipelineStage.DEVELOPMENT,
+                        planning_result["result"]
+                    )
+                    
+                    testing_result = await self.pipeline.execute_stage(
+                        self.pipeline.PipelineStage.TESTING,
+                        development_result["result"]
+                    )
+                    
+                    # Analyze results
+                    quality_analysis = self._analyze_quality(testing_result["result"])
+                    
+                    # Record iteration history
+                    self.history.append({
+                        "iteration": self.current_iteration,
+                        "quality_metrics": quality_analysis,
+                        "timestamp": datetime.now()
+                    })
+                    
+                    # Check if quality requirements are met
+                    if self._meets_quality_requirements(quality_analysis):
+                        self.logger.info("Quality requirements met. Refinement cycle complete.")
+                        return self._prepare_final_result(quality_analysis)
+                    
+                    self.current_iteration += 1
+                
+                return {
+                    "status": "max_iterations_reached",
+                    "iterations_completed": self.current_iteration,
+                    "final_quality": quality_analysis
+                }
+                
+            except Exception as e:
+                self.logger.error(f"Error in refinement cycle: {str(e)}")
+                return {"status": "error", "error": str(e)}
+
+        def _analyze_quality(self, result: Dict[str, Any]) -> Dict[str, float]:
+            """Analyze the quality metrics of the current iteration"""
+            return {
+                "code_quality": self.quality_metrics.code_quality_score,
+                "test_coverage": self.quality_metrics.test_coverage,
+                "security_score": float(self.quality_metrics.security_score)
+            }
+
+        def _meets_quality_requirements(self, quality_analysis: Dict[str, float]) -> bool:
+            """Check if the current quality metrics meet the requirements"""
+            thresholds = self.quality_metrics.thresholds
+            return (
+                quality_analysis["code_quality"] >= thresholds["code_quality"] and
+                quality_analysis["test_coverage"] >= thresholds["test_coverage"] and
+                quality_analysis["security_score"] >= thresholds["security"]
+            )
+
+        def _prepare_final_result(self, quality_analysis: Dict[str, float]) -> Dict[str, Any]:
+            """Prepare the final result of the refinement cycle"""
+            return {
+                "status": "success",
+                "iterations_completed": self.current_iteration,
+                "final_quality": quality_analysis,
+                "history": self.history
+            }
+
+        def get_refinement_history(self) -> List[Dict[str, Any]]:
+            """Get the history of refinement iterations"""
+            return self.history    
+    
+    def run(self):
+        """Main application entry point"""
+        try:
+            logging.info("Starting Autonomous Agent Application")
+            self.interface.render_main_interface()
+        except Exception as e:
+            logging.error(f"Application error: {str(e)}")
+            st.error("An error occurred while starting the application. Please check the logs.")
+            raise
+       
+
     @dataclass
     class QualityMetrics:
         """Advanced quality metrics tracking and analysis"""
@@ -719,4 +723,4 @@ class AutonomousAgentApp:
             st.write(f"Available Memory: {psutil.virtual_memory().available / (1024**3):.1f} GB free")
 
 if __name__ == "__main__":
-    print()
+    app.launch()