Coverage for integrations / coding_agent / goal_manager.py: 95.5%

1""" 

2HevolveSocial - Coding Goal Manager (Unified Delegate) 

3 

4Thin wrapper over the unified GoalManager in agent_engine. 

5All goals are stored as AgentGoal with goal_type='coding'. 

6Keeps backwards-compatible API for coding_agent.api and coding_daemon. 

7""" 

8import re 

9import logging 

10from typing import Dict, List 

11 

12logger = logging.getLogger('hevolve_social') 

13 

14 

15class CodingGoalManager: 

16 """Backwards-compatible API. Delegates to unified GoalManager.""" 

17 

18 @staticmethod 

19 def _sanitize_repo_url(repo_url: str) -> str: 

20 """Validate and sanitize repo_url to prevent command injection.""" 

21 if not repo_url or not re.match(r'^[a-zA-Z0-9_.-]+/[a-zA-Z0-9_.-]+$', repo_url): 

22 raise ValueError(f'Invalid repo_url format: {repo_url}') 

23 if '..' in repo_url: 

24 raise ValueError('repo_url cannot contain path traversal') 

25 return repo_url 

26 

27 @staticmethod 

28 def _sanitize_branch(branch: str) -> str: 

29 """Validate branch name to prevent command injection.""" 

30 if not branch or not re.match(r'^[a-zA-Z0-9_./\\-]+$', branch): 

31 raise ValueError(f'Invalid branch name: {branch}') 

32 if '..' in branch: 

33 raise ValueError('Branch name cannot contain path traversal') 

34 return branch 

35 

36 @staticmethod 

37 def create_goal(db, title: str, description: str, repo_url: str, 

38 branch: str = 'main', target_path: str = '', 

39 created_by: str = None) -> Dict: 

40 """Create a coding goal via the unified GoalManager.""" 

41 from integrations.agent_engine.goal_manager import GoalManager 

42 

43 repo_url = CodingGoalManager._sanitize_repo_url(repo_url) 

44 branch = CodingGoalManager._sanitize_branch(branch) 

45 

46 result = GoalManager.create_goal( 

47 db, goal_type='coding', title=title, description=description, 

48 config={ 

49 'repo_url': repo_url, 

50 'repo_branch': branch, 

51 'target_path': target_path, 

52 }, 

53 created_by=created_by, 

54 ) 

55 return result.get('goal', result) 

56 

57 @staticmethod 

58 def get_goal(db, goal_id: str) -> Dict: 

59 """Get a single goal.""" 

60 from integrations.agent_engine.goal_manager import GoalManager 

61 return GoalManager.get_goal(db, goal_id) 

62 

63 @staticmethod 

64 def update_goal_status(db, goal_id: str, status: str) -> Dict: 

65 """Update goal status.""" 

66 from integrations.agent_engine.goal_manager import GoalManager 

67 return GoalManager.update_goal_status(db, goal_id, status) 

68 

69 @staticmethod 

70 def list_goals(db, status: str = None) -> List[Dict]: 

71 """List coding goals.""" 

72 from integrations.agent_engine.goal_manager import GoalManager 

73 return GoalManager.list_goals(db, goal_type='coding', status=status) 

74 

75 @staticmethod 

76 def build_prompt(goal_dict: dict) -> str: 

77 """Build prompt via the unified GoalManager prompt builder registry.""" 

78 from integrations.agent_engine.goal_manager import GoalManager 

79 

80 # Ensure goal_type is set for the builder lookup 

81 if 'goal_type' not in goal_dict: 

82 goal_dict = dict(goal_dict, goal_type='coding') 

83 return GoalManager.build_prompt(goal_dict)