Coverage for core / platform / app_registry.py: 100.0%

1""" 

2App Registry — Discovery and lifecycle for all HART OS applications. 

3 

4Central catalog of every app in the OS — panels, desktop apps, agents, 

5MCP servers, extensions. Provides search, filtering, and backward 

6compatibility with shell_manifest.py. 

7 

8Integrates with EventBus to emit app.registered / app.unregistered events. 

9 

10Usage: 

11 registry = AppRegistry() 

12 registry.register(AppManifest(id='feed', name='Feed', ...)) 

13 registry.list_by_group('Discover') # -> [AppManifest, ...] 

14 registry.search('remote') # -> [rustdesk, moonlight, ...] 

15 manifest = registry.to_shell_manifest() # backward compat 

16""" 

17 

18import logging 

19import threading 

20from typing import Any, Callable, Dict, List, Optional 

21 

22from core.platform.app_manifest import AppManifest, AppType 

23 

24logger = logging.getLogger('hevolve.platform') 

25 

26 

27class AppRegistry: 

28 """Central app catalog for HART OS. 

29 

30 Stores AppManifest entries and provides query/search capabilities. 

31 Emits events via optional event_emitter callback. 

32 """ 

33 

34 def __init__(self, event_emitter: Optional[Callable] = None): 

35 """Initialize the registry. 

36 

37 Args: 

38 event_emitter: Optional callable(topic, data) for event bus 

39 integration. Called on register/unregister. 

40 """ 

41 self._apps: Dict[str, AppManifest] = {} 

42 self._lock = threading.Lock() 

43 self._emit = event_emitter 

44 

45 def register(self, manifest: AppManifest) -> None: 

46 """Register an app manifest. 

47 

48 Args: 

49 manifest: The app manifest to register. 

50 

51 Raises: 

52 ValueError: If manifest invalid or app ID already registered. 

53 """ 

54 from core.platform.manifest_validator import ManifestValidator 

55 valid, errors = ManifestValidator.validate(manifest) 

56 if not valid: 

57 raise ValueError( 

58 f"Invalid manifest '{manifest.id}': {'; '.join(errors)}") 

59 

60 with self._lock: 

61 if manifest.id in self._apps: 

62 raise ValueError(f"App '{manifest.id}' already registered") 

63 self._apps[manifest.id] = manifest 

64 

65 if self._emit: 

66 self._emit('app.registered', {'app_id': manifest.id, 

67 'name': manifest.name, 

68 'type': manifest.type}) 

69 

70 def unregister(self, app_id: str) -> None: 

71 """Remove an app from the registry. 

72 

73 Args: 

74 app_id: App ID to remove. 

75 

76 Raises: 

77 KeyError: If app ID not found. 

78 """ 

79 with self._lock: 

80 if app_id not in self._apps: 

81 raise KeyError(f"App '{app_id}' not registered") 

82 manifest = self._apps.pop(app_id) 

83 

84 if self._emit: 

85 self._emit('app.unregistered', {'app_id': app_id, 

86 'name': manifest.name}) 

87 

88 def get(self, app_id: str) -> Optional[AppManifest]: 

89 """Get a specific app manifest by ID.""" 

90 return self._apps.get(app_id) 

91 

92 def list_all(self) -> List[AppManifest]: 

93 """Return all registered app manifests.""" 

94 return list(self._apps.values()) 

95 

96 def list_by_type(self, app_type: str) -> List[AppManifest]: 

97 """Return apps matching a specific type. 

98 

99 Args: 

100 app_type: AppType value string (e.g., 'nunba_panel', 'desktop_app'). 

101 """ 

102 return [m for m in self._apps.values() if m.type == app_type] 

103 

104 def list_by_group(self, group: str) -> List[AppManifest]: 

105 """Return apps in a specific start menu group. 

106 

107 Args: 

108 group: Group name (e.g., 'Discover', 'System', 'Remote'). 

109 """ 

110 return [m for m in self._apps.values() 

111 if m.group.lower() == group.lower()] 

112 

113 def list_by_capability(self, capability_type: str) -> List[AppManifest]: 

114 """Return apps that declare a specific AI capability type. 

115 

116 Args: 

117 capability_type: 'llm', 'tts', 'vision', etc. 

118 """ 

119 return [m for m in self._apps.values() 

120 if any(c.get('type') == capability_type 

121 for c in m.ai_capabilities)] 

122 

123 def search(self, query: str) -> List[AppManifest]: 

124 """Fuzzy search across app name, ID, description, tags. 

125 

126 Args: 

127 query: Search string (case-insensitive). 

128 

129 Returns: 

130 Matching manifests, sorted by relevance (exact ID match first). 

131 """ 

132 if not query: 

133 return self.list_all() 

134 

135 results = [m for m in self._apps.values() 

136 if m.matches_search(query)] 

137 

138 # Sort: exact ID match first, then by name 

139 q_lower = query.lower() 

140 results.sort(key=lambda m: ( 

141 0 if m.id.lower() == q_lower else 

142 1 if q_lower in m.id.lower() else 

143 2 if q_lower in m.name.lower() else 3, 

144 m.name.lower(), 

145 )) 

146 return results 

147 

148 def groups(self) -> List[str]: 

149 """Return all unique group names.""" 

150 seen = set() 

151 result = [] 

152 for m in self._apps.values(): 

153 if m.group and m.group not in seen: 

154 seen.add(m.group) 

155 result.append(m.group) 

156 return sorted(result) 

157 

158 def count(self) -> int: 

159 """Return total number of registered apps.""" 

160 return len(self._apps) 

161 

162 # ── Backward Compatibility ──────────────────────────────── 

163 

164 def to_shell_manifest(self) -> Dict[str, Dict[str, Any]]: 

165 """Convert to shell_manifest.py PANEL_MANIFEST format. 

166 

167 For backward compatibility with LiquidUI rendering. 

168 """ 

169 result = {} 

170 for m in self._apps.values(): 

171 if m.type in (AppType.NUNBA_PANEL.value, 

172 AppType.SYSTEM_PANEL.value, 

173 AppType.DYNAMIC_PANEL.value): 

174 result[m.id] = { 

175 'title': m.name, 

176 'icon': m.icon, 

177 'route': m.entry.get('route', ''), 

178 'group': m.group, 

179 'default_size': list(m.default_size), 

180 'apis': m.apis, 

181 } 

182 return result 

183 

184 def load_panel_manifest(self, panels: Dict[str, dict]) -> int: 

185 """Bulk-import from shell_manifest.py PANEL_MANIFEST dict. 

186 

187 Args: 

188 panels: Dict of panel_id -> panel dict from shell_manifest.py. 

189 

190 Returns: 

191 Number of panels imported. 

192 """ 

193 count = 0 

194 for panel_id, panel in panels.items(): 

195 if panel_id not in self._apps: 

196 manifest = AppManifest.from_panel_manifest(panel_id, panel) 

197 with self._lock: 

198 self._apps[manifest.id] = manifest 

199 count += 1 

200 return count 

201 

202 def load_system_panels(self, panels: Dict[str, dict]) -> int: 

203 """Bulk-import from shell_manifest.py SYSTEM_PANELS dict. 

204 

205 Args: 

206 panels: Dict of panel_id -> panel dict. 

207 

208 Returns: 

209 Number of panels imported. 

210 """ 

211 count = 0 

212 for panel_id, panel in panels.items(): 

213 if panel_id not in self._apps: 

214 manifest = AppManifest.from_system_panel(panel_id, panel) 

215 with self._lock: 

216 self._apps[manifest.id] = manifest 

217 count += 1 

218 return count 

219 

220 # ── Lifecycle (for ServiceRegistry) ─────────────────────── 

221 

222 def health(self) -> dict: 

223 """Health report.""" 

224 type_counts = {} 

225 for m in self._apps.values(): 

226 type_counts[m.type] = type_counts.get(m.type, 0) + 1 

227 return { 

228 'status': 'ok', 

229 'total_apps': len(self._apps), 

230 'types': type_counts, 

231 'groups': self.groups(), 

232 }