Coverage for integrations / coding_agent / tool_backends.py: 62.1%

1""" 

2Coding Agent Tool Backends — KiloCode, Claude Code, OpenCode, Aider Native. 

3 

4Subprocess backends wrap CLI tools via subprocess. AiderNativeBackend runs 

5in-process using vendored Aider modules for zero-latency code intelligence. 

6 

7The orchestrator calls exactly ONE backend per task (never all three). 

8This is a leaf tool — never re-dispatches to /chat. 

9""" 

10import json 

11import logging 

12import os 

13import shutil 

14import subprocess 

15import time 

16from abc import ABC, abstractmethod 

17from typing import Dict, List, Optional 

18 

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

20 

21 

22class CodingToolBackend(ABC): 

23 """Base class for coding tool subprocess wrappers.""" 

24 

25 name: str = '' 

26 binary: str = '' 

27 strengths: List[str] = [] 

28 

29 def is_installed(self) -> bool: 

30 return shutil.which(self.binary) is not None 

31 

32 @abstractmethod 

33 def build_command(self, task: str, context: Optional[Dict] = None) -> List[str]: 

34 """Build the CLI command for execution.""" 

35 

36 @abstractmethod 

37 def parse_output(self, stdout: str, stderr: str, returncode: int) -> Dict: 

38 """Parse subprocess output into structured result.""" 

39 

40 def get_capabilities(self) -> Dict: 

41 return { 

42 'name': self.name, 

43 'binary': self.binary, 

44 'installed': self.is_installed(), 

45 'strengths': self.strengths, 

46 } 

47 

48 def get_env(self) -> Dict[str, str]: 

49 """Build environment for subprocess. 

50 

51 For 'own' tasks: passes through all API keys. 

52 For 'hive'/'idle' tasks: strips metered API keys unless the node 

53 operator explicitly opted in via compute policy (fail-closed). 

54 """ 

55 env = os.environ.copy() 

56 task_source = os.environ.get('_CURRENT_TASK_SOURCE', 'own') 

57 allow_metered = True 

58 

59 if task_source in ('hive', 'idle'): 

60 try: 

61 from integrations.agent_engine.compute_config import get_compute_policy 

62 policy = get_compute_policy(os.environ.get('HEVOLVE_NODE_ID')) 

63 allow_metered = policy.get('allow_metered_for_hive', False) 

64 except ImportError: 

65 allow_metered = False # Fail-closed 

66 

67 metered_keys = ('OPENAI_API_KEY', 'ANTHROPIC_API_KEY', 'GROQ_API_KEY', 

68 'GOOGLE_API_KEY', 'OPENROUTER_API_KEY') 

69 for key in metered_keys: 

70 val = os.environ.get(key) 

71 if val and (allow_metered or task_source == 'own'): 

72 env[key] = val 

73 else: 

74 env.pop(key, None) 

75 

76 return env 

77 

78 def execute(self, task: str, context: Optional[Dict] = None, 

79 timeout: int = 300) -> Dict: 

80 """Execute a coding task via subprocess. 

81 

82 This is a TERMINAL operation — calls external CLI process, 

83 never re-dispatches to /chat or creates new agents. 

84 

85 Returns: 

86 {success, output, tool, execution_time_s, error?} 

87 """ 

88 if not self.is_installed(): 

89 return { 

90 'success': False, 

91 'output': '', 

92 'tool': self.name, 

93 'execution_time_s': 0, 

94 'error': f'{self.name} not installed', 

95 } 

96 

97 cmd = self.build_command(task, context) 

98 logger.info(f"Executing {self.name}: {cmd[0]} ...") 

99 

100 start = time.time() 

101 try: 

102 result = subprocess.run( 

103 cmd, 

104 capture_output=True, 

105 text=True, 

106 timeout=timeout, 

107 env=self.get_env(), 

108 cwd=context.get('working_dir') if context else None, 

109 ) 

110 elapsed = time.time() - start 

111 parsed = self.parse_output(result.stdout, result.stderr, result.returncode) 

112 parsed['tool'] = self.name 

113 parsed['execution_time_s'] = round(elapsed, 2) 

114 return parsed 

115 

116 except subprocess.TimeoutExpired: 

117 elapsed = time.time() - start 

118 return { 

119 'success': False, 

120 'output': '', 

121 'tool': self.name, 

122 'execution_time_s': round(elapsed, 2), 

123 'error': f'Timeout after {timeout}s', 

124 } 

125 except (OSError, FileNotFoundError) as e: 

126 return { 

127 'success': False, 

128 'output': '', 

129 'tool': self.name, 

130 'execution_time_s': 0, 

131 'error': str(e), 

132 } 

133 

134 

135class KiloCodeBackend(CodingToolBackend): 

136 """KiloCode CLI wrapper — Apache 2.0 licensed.""" 

137 

138 name = 'kilocode' 

139 binary = 'kilocode' 

140 strengths = ['app_building', 'model_gateway', 'ide_integration', 'multi_provider'] 

141 

142 def build_command(self, task: str, context: Optional[Dict] = None) -> List[str]: 

143 cmd = [self.binary, '--auto', '--json-io'] 

144 if context and context.get('model'): 

145 cmd.extend(['--model', context['model']]) 

146 cmd.extend(['--prompt', task]) 

147 return cmd 

148 

149 def parse_output(self, stdout: str, stderr: str, returncode: int) -> Dict: 

150 try: 

151 data = json.loads(stdout) 

152 return { 

153 'success': returncode == 0, 

154 'output': data.get('result', data.get('output', stdout)), 

155 'metadata': data, 

156 } 

157 except (json.JSONDecodeError, ValueError): 

158 return { 

159 'success': returncode == 0, 

160 'output': stdout or stderr, 

161 } 

162 

163 

164class ClaudeCodeBackend(CodingToolBackend): 

165 """Claude Code CLI wrapper — Proprietary (Anthropic Commercial ToS). 

166 

167 User must install themselves and provide their own ANTHROPIC_API_KEY. 

168 """ 

169 

170 name = 'claude_code' 

171 binary = 'claude' 

172 strengths = ['code_review', 'debugging', 'terminal_workflows', 'complex_reasoning'] 

173 

174 def build_command(self, task: str, context: Optional[Dict] = None) -> List[str]: 

175 cmd = [self.binary, '-p', task, '--output-format', 'json', '--print'] 

176 if context and context.get('model'): 

177 cmd.extend(['--model', context['model']]) 

178 return cmd 

179 

180 def parse_output(self, stdout: str, stderr: str, returncode: int) -> Dict: 

181 try: 

182 data = json.loads(stdout) 

183 # Claude Code JSON output has a 'result' field 

184 output_text = data.get('result', '') 

185 if not output_text and isinstance(data, list): 

186 # Array format: extract text from content blocks 

187 output_text = '\n'.join( 

188 item.get('text', '') for item in data 

189 if isinstance(item, dict) and item.get('type') == 'text' 

190 ) 

191 return { 

192 'success': returncode == 0, 

193 'output': output_text or stdout, 

194 'metadata': data, 

195 } 

196 except (json.JSONDecodeError, ValueError): 

197 return { 

198 'success': returncode == 0, 

199 'output': stdout or stderr, 

200 } 

201 

202 

203class OpenCodeBackend(CodingToolBackend): 

204 """OpenCode CLI wrapper — MIT licensed.""" 

205 

206 name = 'opencode' 

207 binary = 'opencode' 

208 strengths = ['multi_session', 'lsp_integration', 'refactoring', 'session_sharing'] 

209 

210 def build_command(self, task: str, context: Optional[Dict] = None) -> List[str]: 

211 cmd = [self.binary, '-p', task, '-f', 'json'] 

212 if context and context.get('model'): 

213 cmd.extend(['--model', context['model']]) 

214 return cmd 

215 

216 def parse_output(self, stdout: str, stderr: str, returncode: int) -> Dict: 

217 try: 

218 data = json.loads(stdout) 

219 return { 

220 'success': returncode == 0, 

221 'output': data.get('result', data.get('output', stdout)), 

222 'metadata': data, 

223 } 

224 except (json.JSONDecodeError, ValueError): 

225 return { 

226 'success': returncode == 0, 

227 'output': stdout or stderr, 

228 } 

229 

230 

231# Lazy import for AiderNativeBackend to avoid hard dependency 

232def _get_aider_native_class(): 

233 from .aider_native_backend import AiderNativeBackend 

234 return AiderNativeBackend 

235 

236 

237class _LazyAiderNative: 

238 """Lazy proxy so BACKENDS dict doesn't force-import aider_core at module load.""" 

239 

240 _cls = None 

241 

242 def __call__(self): 

243 if self._cls is None: 

244 try: 

245 self._cls = _get_aider_native_class() 

246 except ImportError: 

247 return None 

248 return self._cls() 

249 

250 def __eq__(self, other): 

251 return False # Never matches shutil.which checks 

252 

253 

254# ─── Claw Native Backend (Rust via PyO3, zero subprocess overhead) ─────────── 

255 

256class _LazyClaw: 

257 """Lazy proxy for ClawNativeBackend — avoids hard dep on compiled claw_bridge.""" 

258 

259 _cls = None 

260 

261 def __call__(self): 

262 if self._cls is None: 

263 try: 

264 self._cls = _get_claw_native_class() 

265 except ImportError: 

266 return None 

267 return self._cls() 

268 

269 def __eq__(self, other): 

270 return False 

271 

272 

273def _get_claw_native_class(): 

274 """Import claw_bridge (compiled Rust PyO3 module) and return backend class.""" 

275 import claw_bridge # noqa: F401 — compiled .pyd/.so from claw_native/rust/crates/hart-bridge 

276 

277 class ClawNativeBackend(CodingToolBackend): 

278 """In-process Rust coding agent — bash, file ops, grep, glob via claw-code. 

279 

280 Calls compiled Rust functions directly (no subprocess). Complements the 

281 existing Aider Native backend with Rust-speed file ops and the claw agent 

282 loop for terminal-native coding tasks. 

283 """ 

284 name = 'claw_native' 

285 binary = '' # No subprocess — runs in-process via PyO3 

286 strengths = [ 

287 'terminal_coding', 'file_editing', 'bash_execution', 

288 'grep_search', 'glob_search', 'repo_exploration', 

289 'lsp_integration', 'session_persistence', 

290 ] 

291 

292 def is_installed(self) -> bool: 

293 try: 

294 import claw_bridge 

295 return True 

296 except ImportError: 

297 return False 

298 

299 def build_command(self, task, context=None): 

300 return [] # Not subprocess-based 

301 

302 def parse_output(self, stdout, stderr, returncode): 

303 return {'success': returncode == 0, 'output': stdout, 'error': stderr} 

304 

305 def execute(self, task: str, context=None, timeout: int = 120) -> Dict: 

306 """Execute coding task via Rust claw_bridge — zero subprocess overhead.""" 

307 import json as _json 

308 working_dir = (context or {}).get('working_dir', os.getcwd()) 

309 

310 try: 

311 import shlex 

312 cmd = f'cd {shlex.quote(working_dir)} && {task}' if working_dir else task 

313 result = claw_bridge.execute_bash(cmd, timeout * 1000) 

314 parsed = _json.loads(result) 

315 return { 

316 'success': parsed.get('exit_code', 1) == 0, 

317 'output': parsed.get('stdout', ''), 

318 'error': parsed.get('stderr', ''), 

319 'exit_code': parsed.get('exit_code', 1), 

320 'backend': 'claw_native', 

321 } 

322 except Exception as e: 

323 return {'success': False, 'output': '', 'error': str(e), 'backend': 'claw_native'} 

324 

325 def get_capabilities(self) -> Dict: 

326 caps = super().get_capabilities() 

327 caps['type'] = 'native_rust' 

328 caps['tools'] = ['bash', 'read_file', 'write_file', 'edit_file', 'glob', 'grep'] 

329 return caps 

330 

331 return ClawNativeBackend 

332 

333 

334# Registry of all backends 

335BACKENDS = { 

336 'kilocode': KiloCodeBackend, 

337 'claude_code': ClaudeCodeBackend, 

338 'opencode': OpenCodeBackend, 

339 'aider_native': _LazyAiderNative(), 

340 'claw_native': _LazyClaw(), 

341} 

342 

343 

344def get_available_backends() -> Dict[str, CodingToolBackend]: 

345 """Return instantiated backends for installed tools only.""" 

346 result = {} 

347 for name, cls in BACKENDS.items(): 

348 instance = cls() 

349 if instance is not None and instance.is_installed(): 

350 result[name] = instance 

351 return result 

352 

353 

354def get_all_backends() -> Dict[str, CodingToolBackend]: 

355 """Return all backend instances regardless of installation.""" 

356 result = {} 

357 for name, cls in BACKENDS.items(): 

358 instance = cls() 

359 if instance is not None: 

360 result[name] = instance 

361 return result