Coverage for src / agent / skills / registry.py: 100%

1# Copyright 2025-2026 Microsoft Corporation 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14 

15"""Skill registry for tracking installed skills. 

16 

17This module provides persistence and lookup for installed skills with 

18atomic file operations and canonical name matching. 

19""" 

20 

21import json 

22import os 

23import tempfile 

24from datetime import datetime 

25from pathlib import Path 

26from typing import Any 

27 

28from agent.skills.errors import SkillNotFoundError 

29from agent.skills.manifest import SkillRegistryEntry 

30from agent.skills.security import normalize_skill_name 

31 

32 

33class SkillRegistry: 

34 """Registry for tracking installed skills with JSON persistence. 

35 

36 Provides atomic writes and case-insensitive lookups by canonical name. 

37 

38 Attributes: 

39 registry_path: Path to registry.json file 

40 

41 Example: 

42 >>> registry = SkillRegistry() 

43 >>> entry = SkillRegistryEntry( 

44 ... name="kalshi-markets", 

45 ... name_canonical="kalshi-markets", 

46 ... installed_path=Path("/path/to/skill"), 

47 ... trusted=True 

48 ... ) 

49 >>> registry.register(entry) 

50 >>> skill = registry.get("kalshi-markets") 

51 """ 

52 

53 def __init__(self, registry_path: Path | None = None): 

54 """Initialize skill registry. 

55 

56 Args: 

57 registry_path: Custom registry path (defaults to ~/.osdu-agent/skills/registry.json) 

58 """ 

59 if registry_path is None: 

60 registry_path = Path.home() / ".osdu-agent" / "skills" / "registry.json" 

61 

62 self.registry_path = registry_path 

63 self._ensure_registry_dir() 

64 

65 def _ensure_registry_dir(self) -> None: 

66 """Ensure registry directory exists.""" 

67 self.registry_path.parent.mkdir(parents=True, exist_ok=True) 

68 

69 def _load_registry(self) -> dict[str, Any]: 

70 """Load registry from JSON file. 

71 

72 Returns: 

73 Dictionary mapping canonical names to skill data 

74 """ 

75 if not self.registry_path.exists(): 

76 return {} 

77 

78 try: 

79 with open(self.registry_path, encoding="utf-8") as f: 

80 data: dict[str, Any] = json.load(f) 

81 # Convert ISO datetime strings back to datetime objects 

82 for entry in data.values(): 

83 if "installed_at" in entry: 

84 entry["installed_at"] = datetime.fromisoformat(entry["installed_at"]) 

85 if "installed_path" in entry: 

86 entry["installed_path"] = Path(entry["installed_path"]) 

87 return data 

88 except (json.JSONDecodeError, KeyError): 

89 # Corrupted registry, start fresh 

90 return {} 

91 

92 def _save_registry(self, data: dict[str, Any]) -> None: 

93 """Save registry to JSON file atomically. 

94 

95 Uses temp file + os.replace() for atomic writes to prevent corruption. 

96 

97 Args: 

98 data: Registry data to save 

99 """ 

100 # Convert to JSON-serializable format 

101 serializable = {} 

102 for canonical_name, entry in data.items(): 

103 entry_copy = entry.copy() 

104 if "installed_at" in entry_copy and isinstance(entry_copy["installed_at"], datetime): 

105 entry_copy["installed_at"] = entry_copy["installed_at"].isoformat() 

106 if "installed_path" in entry_copy and isinstance(entry_copy["installed_path"], Path): 

107 entry_copy["installed_path"] = str(entry_copy["installed_path"]) 

108 serializable[canonical_name] = entry_copy 

109 

110 # Atomic write via temp file + os.replace() 

111 fd, temp_path = tempfile.mkstemp( 

112 dir=self.registry_path.parent, prefix=".registry-", suffix=".tmp" 

113 ) 

114 try: 

115 with os.fdopen(fd, "w", encoding="utf-8") as f: 

116 json.dump(serializable, f, indent=2) 

117 # Atomic replace (cross-platform safe) 

118 os.replace(temp_path, self.registry_path) 

119 except Exception: 

120 # Clean up temp file on error 

121 try: 

122 os.unlink(temp_path) 

123 except FileNotFoundError: 

124 # Safe to ignore if temp file doesn't exist (already deleted or never created) 

125 pass 

126 raise 

127 

128 def register(self, entry: SkillRegistryEntry) -> None: 

129 """Register a new skill. 

130 

131 Args: 

132 entry: SkillRegistryEntry to add 

133 

134 Raises: 

135 ValueError: If skill with canonical name already exists 

136 """ 

137 data = self._load_registry() 

138 

139 if entry.name_canonical in data: 

140 raise ValueError(f"Skill '{entry.name_canonical}' already registered") 

141 

142 # Convert to dict for storage 

143 entry_dict = entry.model_dump() 

144 data[entry.name_canonical] = entry_dict 

145 

146 self._save_registry(data) 

147 

148 def unregister(self, canonical_name: str) -> None: 

149 """Unregister a skill. 

150 

151 Args: 

152 canonical_name: Canonical skill name 

153 

154 Raises: 

155 SkillNotFoundError: If skill not found in registry 

156 """ 

157 data = self._load_registry() 

158 

159 if canonical_name not in data: 

160 raise SkillNotFoundError(f"Skill '{canonical_name}' not found in registry") 

161 

162 del data[canonical_name] 

163 self._save_registry(data) 

164 

165 def get(self, name: str) -> SkillRegistryEntry: 

166 """Get skill by name (case-insensitive). 

167 

168 Args: 

169 name: Skill name (any case/format) 

170 

171 Returns: 

172 SkillRegistryEntry for the skill 

173 

174 Raises: 

175 SkillNotFoundError: If skill not found 

176 """ 

177 canonical_name = normalize_skill_name(name) 

178 data = self._load_registry() 

179 

180 if canonical_name not in data: 

181 raise SkillNotFoundError(f"Skill '{name}' not found in registry") 

182 

183 return SkillRegistryEntry(**data[canonical_name]) 

184 

185 def get_by_canonical_name(self, canonical_name: str) -> SkillRegistryEntry: 

186 """Get skill by canonical name (already normalized). 

187 

188 Args: 

189 canonical_name: Canonical skill name (lowercase, hyphens) 

190 

191 Returns: 

192 SkillRegistryEntry for the skill 

193 

194 Raises: 

195 SkillNotFoundError: If skill not found 

196 """ 

197 data = self._load_registry() 

198 

199 if canonical_name not in data: 

200 raise SkillNotFoundError(f"Skill '{canonical_name}' not found in registry") 

201 

202 return SkillRegistryEntry(**data[canonical_name]) 

203 

204 def list(self) -> list[SkillRegistryEntry]: 

205 """List all registered skills. 

206 

207 Returns: 

208 List of SkillRegistryEntry, sorted by canonical name (stable order) 

209 """ 

210 data = self._load_registry() 

211 

212 # Sort by canonical name for stable ordering 

213 sorted_names = sorted(data.keys()) 

214 

215 return [SkillRegistryEntry(**data[name]) for name in sorted_names] 

216 

217 def update_sha(self, canonical_name: str, commit_sha: str) -> None: 

218 """Update commit SHA for a skill. 

219 

220 Args: 

221 canonical_name: Canonical skill name 

222 commit_sha: New commit SHA 

223 

224 Raises: 

225 SkillNotFoundError: If skill not found 

226 """ 

227 data = self._load_registry() 

228 

229 if canonical_name not in data: 

230 raise SkillNotFoundError(f"Skill '{canonical_name}' not found in registry") 

231 

232 data[canonical_name]["commit_sha"] = commit_sha 

233 self._save_registry(data) 

234 

235 def exists(self, name: str) -> bool: 

236 """Check if skill exists in registry. 

237 

238 Args: 

239 name: Skill name (any case/format) 

240 

241 Returns: 

242 True if skill is registered, False otherwise 

243 """ 

244 try: 

245 canonical_name = normalize_skill_name(name) 

246 data = self._load_registry() 

247 return canonical_name in data 

248 except Exception: 

249 return False