Coverage for node / src / stigmem_node / cli / plugins.py: 75%

1"""Plugin inspection CLI handlers.""" 

2 

3from __future__ import annotations 

4 

5import argparse 

6import os 

7from typing import Any 

8 

9PLUGIN_DOCS_URL = "https://docs.stigmem.dev/plugins" 

10 

11KNOWN_PLUGINS: tuple[dict[str, str], ...] = ( 

12 { 

13 "slug": "lazy-instruction-discovery", 

14 "package": "stigmem-plugin-lazy-instruction-discovery", 

15 "env_var": "STIGMEM_LAZY_INSTRUCTION_DISCOVERY_ENABLED", 

16 "summary": "Opt-in instruction manifest discovery and migration helpers.", 

17 }, 

18 { 

19 "slug": "time-travel", 

20 "package": "stigmem-plugin-time-travel", 

21 "env_var": "STIGMEM_TIME_TRAVEL_ENABLED", 

22 "summary": "Opt-in historical fact and recall query behavior.", 

23 }, 

24 { 

25 "slug": "tombstones", 

26 "package": "stigmem-plugin-tombstones", 

27 "env_var": "STIGMEM_TOMBSTONES_ENABLED", 

28 "summary": "Opt-in right-to-be-forgotten tombstone enforcement.", 

29 }, 

30 { 

31 "slug": "memory-garden-acl", 

32 "package": "stigmem-plugin-memory-garden-acl", 

33 "env_var": "STIGMEM_MEMORY_GARDEN_ACL_ENABLED", 

34 "summary": "Opt-in memory garden membership ACL filtering.", 

35 }, 

36 { 

37 "slug": "source-attestation", 

38 "package": "stigmem-plugin-source-attestation", 

39 "env_var": "STIGMEM_SOURCE_ATTESTATION_ENABLED", 

40 "summary": "Opt-in source identity checks and source-trust recall signals.", 

41 }, 

42 { 

43 "slug": "multi-tenant", 

44 "package": "stigmem-plugin-multi-tenant", 

45 "env_var": "STIGMEM_MULTI_TENANT_ENABLED", 

46 "summary": "Opt-in tenant scoping and default-tenant collapse.", 

47 }, 

48) 

49 

50 

51def _load_plugin_registry() -> Any: 

52 from ..plugins import HookRegistry, register_discovered_plugins 

53 

54 registry = HookRegistry() 

55 register_discovered_plugins(registry=registry) 

56 registry.poll_plugin_health() 

57 return registry 

58 

59 

60def _plugin_report_by_name(registry: Any) -> dict[str, Any]: 

61 return {report.plugin_name: report for report in registry.plugin_health_reports()} 

62 

63 

64def _plugin_info_to_dict(info: Any, health: Any | None = None) -> dict[str, Any]: 

65 known = _known_plugin(info.name) 

66 data = { 

67 "name": info.name, 

68 "version": info.version, 

69 "capabilities": list(info.capabilities), 

70 "hooks": list(info.hooks), 

71 "hook_count": len(info.hooks), 

72 "depends_on": list(info.depends_on), 

73 "discovery_source": info.discovery_source, 

74 "signed_by": info.signed_by, 

75 } 

76 if known is not None: 76 ↛ 77line 76 didn't jump to line 77 because the condition on line 76 was never true

77 data["enabled"] = _env_enabled(known["env_var"]) 

78 data["enable_env_var"] = known["env_var"] 

79 if health is not None: 79 ↛ 86line 79 didn't jump to line 86 because the condition on line 79 was always true

80 data["health"] = { 

81 "status": health.status.value, 

82 "message": health.message, 

83 "checked_at": health.checked_at.isoformat(), 

84 "error_summary": health.error_summary, 

85 } 

86 return data 

87 

88 

89def _env_enabled(name: str) -> bool: 

90 return os.getenv(name, "").strip().lower() in {"1", "true", "yes", "on"} 

91 

92 

93def _known_plugin(name: str) -> dict[str, str] | None: 

94 normalized = name.removeprefix("stigmem-plugin-") 

95 for plugin in KNOWN_PLUGINS: 

96 if name in {plugin["slug"], plugin["package"]} or normalized == plugin["slug"]: 

97 return plugin 

98 return None 

99 

100 

101def _installed_plugin_names(registry: Any) -> set[str]: 

102 names: set[str] = set() 

103 for info in registry.plugin_infos(): 103 ↛ 104line 103 didn't jump to line 104 because the loop on line 103 never started

104 names.add(str(info.name)) 

105 known = _known_plugin(str(info.name)) 

106 if known is not None: 

107 names.add(known["slug"]) 

108 names.add(known["package"]) 

109 return names 

110 

111 

112def _plugin_doctor_rows(registry: Any) -> list[dict[str, Any]]: 

113 installed = _installed_plugin_names(registry) 

114 rows: list[dict[str, Any]] = [] 

115 for plugin in KNOWN_PLUGINS: 

116 is_installed = plugin["slug"] in installed or plugin["package"] in installed 

117 is_enabled = _env_enabled(plugin["env_var"]) 

118 if is_enabled and not is_installed: 

119 status = "enabled-not-installed" 

120 recommendation = f"Install {plugin['package']} or unset {plugin['env_var']}." 

121 elif is_installed and not is_enabled: 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true

122 status = "installed-disabled" 

123 recommendation = f"Set {plugin['env_var']}=1 to enable it." 

124 elif is_installed: 124 ↛ 125line 124 didn't jump to line 125 because the condition on line 124 was never true

125 status = "installed-enabled" 

126 recommendation = "No action required." 

127 else: 

128 status = "not-installed" 

129 recommendation = "No action required." 

130 rows.append( 

131 { 

132 "slug": plugin["slug"], 

133 "package": plugin["package"], 

134 "env_var": plugin["env_var"], 

135 "installed": is_installed, 

136 "enabled": is_enabled, 

137 "status": status, 

138 "recommendation": recommendation, 

139 } 

140 ) 

141 return rows 

142 

143 

144def _print_plugin_doctor_rows(rows: list[dict[str, Any]]) -> None: 

145 print("Plugin state diagnostics:") 

146 for row in rows: 

147 print( 

148 f"{row['package']} status={row['status']} " 

149 f"enabled={str(row['enabled']).lower()} env={row['env_var']}" 

150 ) 

151 if row["status"] in {"enabled-not-installed", "installed-disabled"}: 

152 print(f" recommendation: {row['recommendation']}") 

153 if all(row["status"] in {"installed-enabled", "not-installed"} for row in rows): 153 ↛ 154line 153 didn't jump to line 154 because the condition on line 153 was never true

154 print("Plugin configuration looks consistent.") 

155 

156 

157def _cmd_plugins_list(args: argparse.Namespace) -> int: 

158 import json 

159 

160 registry = _load_plugin_registry() 

161 infos = registry.plugin_infos() 

162 health_by_name = _plugin_report_by_name(registry) 

163 rows = [_plugin_info_to_dict(info, health_by_name.get(info.name)) for info in infos] 

164 if args.json: 

165 print(json.dumps(rows, indent=2)) 

166 return 0 

167 if not rows: 167 ↛ 170line 167 didn't jump to line 170 because the condition on line 167 was always true

168 print(f"No plugins registered. See {PLUGIN_DOCS_URL} for the plugin catalog.") 

169 return 0 

170 for row in rows: 

171 health = row.get("health") or {} 

172 health_status = health.get("status", "unknown") 

173 enabled = row.get("enabled") 

174 enabled_text = f" enabled={str(enabled).lower()}" if enabled is not None else "" 

175 print( 

176 f"{row['name']} {row['version']} " 

177 f"hooks={row['hook_count']}{enabled_text} " 

178 f"health={health_status} signed_by={row['signed_by']}" 

179 ) 

180 return 0 

181 

182 

183def _cmd_plugins_describe(args: argparse.Namespace) -> int: 

184 import json 

185 import sys 

186 

187 registry = _load_plugin_registry() 

188 info = registry.plugin_info(args.name) 

189 if info is None: 

190 print(f"error: plugin not found: {args.name}", file=sys.stderr) 

191 return 1 

192 health = _plugin_report_by_name(registry).get(info.name) 

193 data = _plugin_info_to_dict(info, health) 

194 if args.json: 194 ↛ 195line 194 didn't jump to line 195 because the condition on line 194 was never true

195 print(json.dumps(data, indent=2)) 

196 return 0 

197 print(f"name: {data['name']}") 

198 print(f"version: {data['version']}") 

199 print(f"signed_by: {data['signed_by']}") 

200 print(f"capabilities: {', '.join(data['capabilities']) or '-'}") 

201 print(f"hooks ({data['hook_count']}): {', '.join(data['hooks']) or '-'}") 

202 print(f"depends_on: {', '.join(data['depends_on']) or '-'}") 

203 print(f"discovery_source: {json.dumps(data['discovery_source'], sort_keys=True)}") 

204 health_data = data.get("health") or {} 

205 print(f"health: {health_data.get('status', 'unknown')}") 

206 if health_data.get("message"): 206 ↛ 208line 206 didn't jump to line 208 because the condition on line 206 was always true

207 print(f"health_message: {health_data['message']}") 

208 if health_data.get("error_summary"): 208 ↛ 209line 208 didn't jump to line 209 because the condition on line 208 was never true

209 print(f"health_error: {health_data['error_summary']}") 

210 if health_data.get("checked_at"): 210 ↛ 212line 210 didn't jump to line 212 because the condition on line 210 was always true

211 print(f"health_checked_at: {health_data['checked_at']}") 

212 return 0 

213 

214 

215def _cmd_plugins_search(args: argparse.Namespace) -> int: 

216 import json 

217 

218 query = args.query.strip().lower() 

219 rows = [ 

220 plugin 

221 for plugin in KNOWN_PLUGINS 

222 if query in plugin["slug"] 

223 or query in plugin["package"] 

224 or query in plugin["summary"].lower() 

225 ] 

226 if args.json: 226 ↛ 227line 226 didn't jump to line 227 because the condition on line 226 was never true

227 print(json.dumps(rows, indent=2)) 

228 return 0 

229 if not rows: 229 ↛ 230line 229 didn't jump to line 230 because the condition on line 229 was never true

230 print(f"No catalog matches for {args.query!r}. See {PLUGIN_DOCS_URL}.") 

231 return 0 

232 for row in rows: 

233 print(f"{row['package']} - {row['summary']}") 

234 print(f" install: python -m pip install '{row['package']}>=0.1.0,<2.0.0'") 

235 print(f" enable: export {row['env_var']}=1") 

236 return 0 

237 

238 

239def _cmd_plugins_enable(args: argparse.Namespace) -> int: 

240 plugin = _known_plugin(args.name) 

241 if plugin is None: 241 ↛ 242line 241 didn't jump to line 242 because the condition on line 241 was never true

242 print(f"error: unknown plugin: {args.name}") 

243 return 1 

244 print(f"python -m pip install '{plugin['package']}>=0.1.0,<2.0.0'") 

245 print(f"export {plugin['env_var']}=1") 

246 print("Restart the stigmem node after changing plugin packages or gates.") 

247 return 0 

248 

249 

250def _cmd_plugins_disable(args: argparse.Namespace) -> int: 

251 plugin = _known_plugin(args.name) 

252 if plugin is None: 252 ↛ 253line 252 didn't jump to line 253 because the condition on line 252 was never true

253 print(f"error: unknown plugin: {args.name}") 

254 return 1 

255 print(f"unset {plugin['env_var']}") 

256 print("Restart the stigmem node after changing plugin gates.") 

257 return 0 

258 

259 

260def _cmd_plugins_doctor(args: argparse.Namespace) -> int: 

261 import json 

262 

263 registry = _load_plugin_registry() 

264 rows = _plugin_doctor_rows(registry) 

265 if args.json: 265 ↛ 266line 265 didn't jump to line 266 because the condition on line 265 was never true

266 print(json.dumps(rows, indent=2)) 

267 return 0 

268 _print_plugin_doctor_rows(rows) 

269 return 0 

270 

271 

272def _cmd_doctor(args: argparse.Namespace) -> int: 

273 import json 

274 

275 registry = _load_plugin_registry() 

276 plugin_rows = _plugin_doctor_rows(registry) 

277 payload = { 

278 "status": "ok", 

279 "plugins": plugin_rows, 

280 } 

281 if args.json: 281 ↛ 284line 281 didn't jump to line 284 because the condition on line 281 was always true

282 print(json.dumps(payload, indent=2)) 

283 return 0 

284 print("status: ok") 

285 _print_plugin_doctor_rows(plugin_rows) 

286 return 0