Coverage for integrations / openclaw / skill_exporter.py: 81.9%

1""" 

2Skill Exporter — Export HART recipes as ClawHub-compatible SKILL.md files. 

3 

4This is the reverse direction: HART OS recipes become OpenClaw skills, 

5publishable to ClawHub for the 3,200+ skill ecosystem to use. 

6 

7Any trained HART recipe can be exported: 

8 recipe → SKILL.md + supporting files → clawhub publish 

9""" 

10 

11import json 

12import logging 

13import os 

14from pathlib import Path 

15from typing import Any, Dict, List, Optional 

16 

17logger = logging.getLogger(__name__) 

18 

19 

20def recipe_to_skill_md(recipe_path: str, 

21 name: Optional[str] = None, 

22 description: Optional[str] = None, 

23 version: str = '1.0.0') -> str: 

24 """Convert a HART recipe JSON to a ClawHub SKILL.md. 

25 

26 Args: 

27 recipe_path: Path to the recipe JSON file 

28 name: Skill name (defaults to recipe prompt) 

29 description: Skill description 

30 version: Semver version 

31 

32 Returns: 

33 SKILL.md content string 

34 """ 

35 with open(recipe_path, 'r', encoding='utf-8') as f: 

36 recipe = json.load(f) 

37 

38 # Extract recipe metadata 

39 prompt = recipe.get('prompt', recipe.get('task', '')) 

40 actions = recipe.get('actions', recipe.get('steps', [])) 

41 persona = recipe.get('persona', '') 

42 

43 if not name: 

44 # Generate name from prompt 

45 name = prompt.lower().replace(' ', '-')[:40] 

46 name = ''.join(c for c in name if c.isalnum() or c == '-') 

47 name = f"hart-{name}" 

48 

49 if not description: 

50 description = f"HART OS trained recipe: {prompt[:100]}" 

51 

52 # Build frontmatter 

53 metadata = { 

54 'openclaw': { 

55 'emoji': '\U0001f916', 

56 'requires': { 

57 'env': ['OPENAI_API_KEY'], 

58 }, 

59 'primaryEnv': 'OPENAI_API_KEY', 

60 } 

61 } 

62 

63 lines = [ 

64 '---', 

65 f'name: {name}', 

66 f'description: {description}', 

67 f'version: {version}', 

68 f'homepage: https://github.com/hertz-ai/HARTOS', 

69 f'metadata: {json.dumps(metadata)}', 

70 '---', 

71 '', 

72 f'# {name}', 

73 '', 

74 f'{description}', 

75 '', 

76 '## Instructions', 

77 '', 

78 ] 

79 

80 if persona: 

81 lines.append(f'You are acting as: **{persona}**') 

82 lines.append('') 

83 

84 lines.append(f'Original task: {prompt}') 

85 lines.append('') 

86 

87 # Convert recipe actions to step-by-step instructions 

88 if actions: 

89 lines.append('## Steps') 

90 lines.append('') 

91 for i, action in enumerate(actions, 1): 

92 if isinstance(action, dict): 

93 action_desc = action.get('action', action.get('description', '')) 

94 tool = action.get('tool', action.get('action_type', '')) 

95 expected = action.get('expected_output', '') 

96 lines.append(f'{i}. **{action_desc}**') 

97 if tool: 

98 lines.append(f' - Tool: `{tool}`') 

99 if expected: 

100 lines.append(f' - Expected: {expected}') 

101 else: 

102 lines.append(f'{i}. {action}') 

103 lines.append('') 

104 

105 lines.extend([ 

106 '## Source', 

107 '', 

108 'This skill was generated from a trained HART OS recipe.', 

109 'It can be replayed without LLM calls using HART REUSE mode.', 

110 '', 

111 '---', 

112 '*Exported by HART OS Recipe-to-Skill bridge*', 

113 ]) 

114 

115 return '\n'.join(lines) 

116 

117 

118def export_recipe_as_skill(recipe_path: str, 

119 output_dir: str, 

120 name: Optional[str] = None, 

121 description: Optional[str] = None, 

122 version: str = '1.0.0') -> str: 

123 """Export a HART recipe as a complete ClawHub skill directory. 

124 

125 Args: 

126 recipe_path: Path to the recipe JSON 

127 output_dir: Directory to create the skill in 

128 name: Optional skill name 

129 description: Optional description 

130 version: Semver version 

131 

132 Returns: 

133 Path to the created skill directory 

134 """ 

135 skill_md = recipe_to_skill_md(recipe_path, name, description, version) 

136 

137 out = Path(output_dir) 

138 out.mkdir(parents=True, exist_ok=True) 

139 

140 (out / 'SKILL.md').write_text(skill_md, encoding='utf-8') 

141 

142 # Copy the original recipe as reference 

143 recipe_dest = out / 'hart_recipe.json' 

144 with open(recipe_path, 'r', encoding='utf-8') as f: 

145 recipe_data = json.load(f) 

146 with open(recipe_dest, 'w', encoding='utf-8') as f: 

147 json.dump(recipe_data, f, indent=2) 

148 

149 logger.info("Exported recipe %s as skill at %s", recipe_path, out) 

150 return str(out) 

151 

152 

153def _broadcast_skill_via_p2p(skill_dir: str, slug: Optional[str] = None): 

154 """Broadcast skill to the hive via MessageBus federation.recipe_delta. 

155 

156 This feeds into FederatedAggregator.receive_recipe_delta() so that 

157 exported skills become discoverable by all hive nodes without ClawHub. 

158 """ 

159 try: 

160 skill_path = Path(skill_dir) 

161 recipe_file = skill_path / 'hart_recipe.json' 

162 skill_md_file = skill_path / 'SKILL.md' 

163 

164 # Extract metadata from the recipe JSON (written by export_recipe_as_skill) 

165 recipe_data = {} 

166 if recipe_file.exists(): 

167 with open(recipe_file, 'r', encoding='utf-8') as f: 

168 recipe_data = json.load(f) 

169 

170 # Extract skill name from SKILL.md frontmatter 

171 skill_name = slug or '' 

172 if not skill_name and skill_md_file.exists(): 

173 for line in skill_md_file.read_text(encoding='utf-8').splitlines(): 

174 if line.startswith('name:'): 

175 skill_name = line.split(':', 1)[1].strip() 

176 break 

177 

178 actions = recipe_data.get('actions', recipe_data.get('steps', [])) 

179 skill_id = slug or skill_name or os.path.basename(skill_dir) 

180 

181 delta = { 

182 'recipes': [{ 

183 'id': skill_id, 

184 'name': skill_name, 

185 'action_count': len(actions), 

186 'success_rate': 1.0, 

187 'reuse_count': 0, 

188 }], 

189 } 

190 

191 from core.peer_link.message_bus import get_message_bus 

192 bus = get_message_bus() 

193 bus.publish('federation.recipe_delta', delta) 

194 logger.info("Broadcast skill %s via P2P federation", skill_id) 

195 except Exception as e: 

196 logger.debug("P2P skill broadcast skipped: %s", e) 

197 

198 

199def publish_skill(skill_dir: str, slug: Optional[str] = None) -> bool: 

200 """Publish a skill to ClawHub (requires clawhub CLI). 

201 

202 Also broadcasts the skill via P2P MessageBus so hive nodes can 

203 discover it without relying on external ClawHub infrastructure. 

204 

205 Args: 

206 skill_dir: Path to the skill directory containing SKILL.md 

207 slug: Optional slug override 

208 

209 Returns: 

210 True if published successfully to ClawHub 

211 """ 

212 import shutil 

213 import subprocess 

214 

215 # Always broadcast via P2P regardless of ClawHub availability 

216 _broadcast_skill_via_p2p(skill_dir, slug) 

217 

218 clawhub = shutil.which('clawhub') 

219 if not clawhub: 

220 logger.error("clawhub CLI not installed — cannot publish") 

221 return False 

222 

223 cmd = [clawhub, 'publish', skill_dir] 

224 if slug: 

225 cmd.extend(['--slug', slug]) 

226 

227 try: 

228 result = subprocess.run(cmd, capture_output=True, text=True, timeout=30) 

229 if result.returncode == 0: 

230 logger.info("Published skill from %s", skill_dir) 

231 return True 

232 logger.error("Publish failed: %s", result.stderr) 

233 return False 

234 except Exception as e: 

235 logger.error("Publish error: %s", e) 

236 return False