Coverage for integrations / coding_agent / recipe_bridge.py: 83.1%

1""" 

2Coding Recipe Bridge — Connects Aider edit results to the HARTOS Recipe Pattern. 

3 

4CREATE mode: Captures file edits as recipe steps with search/replace metadata 

5REUSE mode: Replays coding recipes by applying patches directly (no LLM needed) 

6 

7Falls back to full LLM execution if files have changed since recipe capture. 

8""" 

9import json 

10import logging 

11import os 

12from typing import Dict, List, Optional 

13 

14logger = logging.getLogger('hevolve.coding_agent.recipe_bridge') 

15 

16 

17class CodingRecipeBridge: 

18 """Bridges coding tool outputs to HARTOS recipe format.""" 

19 

20 @staticmethod 

21 def capture_edit_as_recipe_step( 

22 task: str, 

23 tool_name: str, 

24 file_edits: List[Dict], 

25 working_dir: str = '', 

26 ) -> Dict: 

27 """Convert coding tool file edits to a HARTOS recipe step. 

28 

29 Args: 

30 task: The original task description 

31 tool_name: Which coding tool produced the edits 

32 file_edits: List of {filename, original, updated} or 

33 {filename, search, replace} dicts 

34 working_dir: Working directory context 

35 

36 Returns: 

37 Recipe step dict compatible with HARTOS recipe format. 

38 """ 

39 # Normalize edits to search/replace blocks 

40 sr_blocks = [] 

41 for edit in file_edits: 

42 filename = edit.get('filename', '') 

43 if 'search' in edit and 'replace' in edit: 

44 sr_blocks.append({ 

45 'filename': filename, 

46 'search': edit['search'], 

47 'replace': edit['replace'], 

48 }) 

49 elif 'original' in edit and 'updated' in edit: 

50 sr_blocks.append({ 

51 'filename': filename, 

52 'search': edit['original'], 

53 'replace': edit['updated'], 

54 }) 

55 

56 return { 

57 'description': task, 

58 'tool_name': tool_name, 

59 'aider_edit_format': 'search_replace', 

60 'search_replace_blocks': sr_blocks, 

61 'working_dir': working_dir, 

62 'files_modified': list({b['filename'] for b in sr_blocks}), 

63 } 

64 

65 @staticmethod 

66 def replay_recipe_step( 

67 step: Dict, 

68 working_dir: str = '', 

69 ) -> Dict: 

70 """Replay a coding recipe step without LLM — apply search/replace directly. 

71 

72 Args: 

73 step: Recipe step dict (must have aider_edit_format + search_replace_blocks) 

74 working_dir: Override working directory 

75 

76 Returns: 

77 {success, files_modified, errors} 

78 """ 

79 if step.get('aider_edit_format') != 'search_replace': 

80 return { 

81 'success': False, 

82 'error': 'Not a coding recipe step (missing aider_edit_format)', 

83 } 

84 

85 blocks = step.get('search_replace_blocks', []) 

86 if not blocks: 

87 return {'success': True, 'files_modified': [], 'errors': []} 

88 

89 base_dir = working_dir or step.get('working_dir', '') 

90 modified = [] 

91 errors = [] 

92 

93 for block in blocks: 

94 filename = block.get('filename', '') 

95 search = block.get('search', '') 

96 replace = block.get('replace', '') 

97 

98 if not filename or not search: 

99 continue 

100 

101 filepath = os.path.join(base_dir, filename) if base_dir else filename 

102 

103 if not os.path.exists(filepath): 

104 errors.append(f'File not found: {filepath}') 

105 continue 

106 

107 try: 

108 with open(filepath, 'r', encoding='utf-8') as f: 

109 content = f.read() 

110 

111 if search not in content: 

112 # Try flexible matching 

113 matched = _flexible_patch(content, search, replace) 

114 if matched is None: 

115 errors.append( 

116 f'Search text not found in {filename} (file may have changed)') 

117 continue 

118 content = matched 

119 else: 

120 content = content.replace(search, replace, 1) 

121 

122 with open(filepath, 'w', encoding='utf-8') as f: 

123 f.write(content) 

124 

125 modified.append(filename) 

126 logger.info(f'Recipe replay: patched {filename}') 

127 

128 except Exception as e: 

129 errors.append(f'Error patching {filename}: {e}') 

130 

131 return { 

132 'success': len(errors) == 0, 

133 'files_modified': modified, 

134 'errors': errors, 

135 } 

136 

137 @staticmethod 

138 def get_repository_map( 

139 working_dir: str = '.', 

140 max_tokens: int = 2048, 

141 ) -> str: 

142 """Generate a tree-sitter repository map for coding context. 

143 

144 Registered as an AutoGen tool so agents can understand repo structure. 

145 """ 

146 try: 

147 from integrations.coding_agent.aider_native_backend import AiderNativeBackend 

148 backend = AiderNativeBackend() 

149 if not backend.is_installed(): 

150 return 'Repository map not available (aider_core not installed)' 

151 return backend.get_repo_map(working_dir=working_dir, max_tokens=max_tokens) or '' 

152 except Exception as e: 

153 return f'Repository map error: {e}' 

154 

155 

156def _flexible_patch(content: str, search: str, replace: str) -> Optional[str]: 

157 """Try flexible search/replace when exact match fails. 

158 

159 Uses the vendored Aider search_replace if available, otherwise 

160 falls back to whitespace-normalized matching. 

161 """ 

162 try: 

163 from integrations.coding_agent.aider_core.coders.search_replace import ( 

164 flexible_search_and_replace, editblock_strategies, 

165 ) 

166 texts = (search, replace, content) 

167 result = flexible_search_and_replace(texts, editblock_strategies) 

168 return result if result != content else None 

169 except ImportError: 

170 pass 

171 

172 # Fallback: normalize whitespace and try again 

173 import re 

174 search_normalized = re.sub(r'\s+', r'\\s+', re.escape(search.strip())) 

175 match = re.search(search_normalized, content) 

176 if match: 

177 return content[:match.start()] + replace + content[match.end():] 

178 

179 return None