Coverage for integrations / coding_agent / tool_router.py: 76.6%

1""" 

2Coding Agent Tool Router — Intelligent routing to best coding tool. 

3 

4Priority order: 

5 1. User override (explicit tool choice) 

6 2. Local benchmark data (min 5 samples per task type) 

7 3. Hive-aggregated intelligence (from FederatedAggregator) 

8 4. Heuristic defaults 

9""" 

10import logging 

11from typing import Optional 

12 

13from .tool_backends import CodingToolBackend, get_available_backends, get_all_backends 

14 

15logger = logging.getLogger('hevolve.coding_agent') 

16 

17# Heuristic defaults when no benchmark data exists. 

18# aider_native preferred for tasks where in-process code intelligence excels. 

19# claw_native preferred for Rust-speed file ops, bash, grep/glob (no subprocess). 

20# Falls through to subprocess backends if native backends aren't available. 

21HEURISTIC_DEFAULTS = { 

22 'code_review': 'claude_code', 

23 'debugging': 'claude_code', 

24 'complex_reasoning': 'claude_code', 

25 'terminal_workflows': 'claw_native', # Rust-native bash + file ops 

26 'terminal_coding': 'claw_native', # Agent loop with tools 

27 'app_build': 'kilocode', 

28 'feature': 'kilocode', 

29 'refactor': 'aider_native', 

30 'bug_fix': 'aider_native', 

31 'multi_session': 'opencode', 

32 'multi_file_edit': 'aider_native', 

33 'architecture': 'claude_code', 

34 'repo_exploration': 'claw_native', # Fast grep/glob via Rust 

35 'bash_execution': 'claw_native', # Sandboxed bash via Rust 

36} 

37 

38 

39class CodingToolRouter: 

40 """Route coding tasks to the best available tool.""" 

41 

42 def route(self, task: str, task_type: str = 'feature', 

43 user_override: str = '', 

44 context: Optional[dict] = None) -> Optional[CodingToolBackend]: 

45 """Select the best backend for this task. 

46 

47 Returns None if no tools are installed. 

48 """ 

49 available = get_available_backends() 

50 if not available: 

51 logger.warning("No coding tools installed") 

52 return None 

53 

54 # 1. User override — respect explicitly 

55 if user_override and user_override in available: 

56 logger.info(f"Router: user override → {user_override}") 

57 return available[user_override] 

58 

59 # 2. Local benchmark data 

60 best = self._check_local_benchmarks(task_type, available) 

61 if best: 

62 logger.info(f"Router: local benchmark → {best.name}") 

63 return best 

64 

65 # 3. Hive-aggregated intelligence 

66 best = self._check_hive_intelligence(task_type, available) 

67 if best: 

68 logger.info(f"Router: hive intelligence → {best.name}") 

69 return best 

70 

71 # 4. Heuristic default 

72 default_name = HEURISTIC_DEFAULTS.get(task_type, '') 

73 if default_name in available: 

74 logger.info(f"Router: heuristic → {default_name}") 

75 return available[default_name] 

76 

77 # 5. First available tool 

78 first = next(iter(available.values())) 

79 logger.info(f"Router: fallback → {first.name}") 

80 return first 

81 

82 def _check_local_benchmarks(self, task_type: str, 

83 available: dict) -> Optional[CodingToolBackend]: 

84 """Check local benchmark DB for best tool.""" 

85 try: 

86 from .benchmark_tracker import get_benchmark_tracker 

87 tracker = get_benchmark_tracker() 

88 result = tracker.get_best_tool(task_type) 

89 if result: 

90 tool_name, success_rate, avg_time = result 

91 if tool_name in available: 

92 return available[tool_name] 

93 except Exception: 

94 pass 

95 return None 

96 

97 def _check_hive_intelligence(self, task_type: str, 

98 available: dict) -> Optional[CodingToolBackend]: 

99 """Check hive-aggregated routing table.""" 

100 try: 

101 from .benchmark_tracker import get_benchmark_tracker 

102 tracker = get_benchmark_tracker() 

103 tool_name = tracker.get_hive_best_tool(task_type) 

104 if tool_name and tool_name in available: 

105 return available[tool_name] 

106 except Exception: 

107 pass 

108 return None