| 1 | from __future__ import annotations |
| 2 | |
| 3 | from pathlib import Path |
| 4 | from typing import Any |
| 5 | |
| 6 | from agent import AgentContext |
| 7 | from helpers import files, projects, skills |
| 8 | from helpers.api import ApiHandler, Request, Response |
| 9 | from helpers.persist_chat import save_tmp_chat |
| 10 | |
| 11 | |
| 12 | class SkillsCatalog(ApiHandler): |
| 13 | async def process(self, input: dict, request: Request) -> dict | Response: |
| 14 | action = str(input.get("action", "list") or "list").strip().lower() |
| 15 | context_id = str(input.get("context_id", "") or "").strip() |
| 16 | project_name = str(input.get("project_name", "") or "").strip() |
| 17 | |
| 18 | try: |
| 19 | if action == "list": |
| 20 | return self._build_state(context_id=context_id, project_name=project_name) |
| 21 | if action == "activate": |
| 22 | return self._activate(input, context_id=context_id) |
| 23 | if action == "deactivate": |
| 24 | return self._deactivate(input, context_id=context_id) |
| 25 | if action == "hide": |
| 26 | return self._hide(input, context_id=context_id) |
| 27 | if action == "show": |
| 28 | return self._show(input, context_id=context_id) |
| 29 | if action == "clear": |
| 30 | return self._clear(context_id=context_id) |
| 31 | if action == "get_doc": |
| 32 | return self._get_doc(input, context_id=context_id, project_name=project_name) |
| 33 | return {"ok": False, "error": f"Unknown action: {action}"} |
| 34 | except Exception as e: |
| 35 | return {"ok": False, "error": str(e)} |
| 36 | |
| 37 | def _activate(self, input: dict, *, context_id: str) -> dict[str, Any]: |
| 38 | context = self._require_context(context_id) |
| 39 | skill_entry = self._require_skill_entry(input) |
| 40 | agent = context.get_agent() |
| 41 | skill = self._resolve_catalog_skill(skill_entry, context=context) |
| 42 | skill_name = str(skill.get("name") or skill_entry.get("name") or "").strip() |
| 43 | if not skill_name: |
| 44 | raise ValueError("Skill name is required") |
| 45 | |
| 46 | skill_path = str(skill.get("path") or skill_entry.get("path") or "").strip() |
| 47 | skills.add_loaded_skill_name(agent, skill_name) |
| 48 | if not self._visible_skill_loaded(agent, skill_name): |
| 49 | content = skills.load_skill_for_agent(skill_name=skill_name, agent=agent) |
| 50 | if content.startswith("Error:"): |
| 51 | raise ValueError(content) |
| 52 | agent.hist_add_tool_result( |
| 53 | "skills_tool", |
| 54 | content, |
| 55 | skill_instructions={ |
| 56 | "name": skill_name, |
| 57 | "path": skill_path, |
| 58 | "source": "skills_page:load", |
| 59 | "content_included": True, |
| 60 | }, |
| 61 | ) |
| 62 | save_tmp_chat(context) |
| 63 | return self._build_state(context_id=context.id) |
| 64 | |
| 65 | def _deactivate(self, input: dict, *, context_id: str) -> dict[str, Any]: |
| 66 | return { |
| 67 | "ok": False, |
| 68 | "error": "Loaded skills are kept in chat history and cannot be removed.", |
| 69 | } |
| 70 | |
| 71 | def _hide(self, input: dict, *, context_id: str) -> dict[str, Any]: |
| 72 | context = self._require_context(context_id) |
| 73 | skill_entry = self._require_skill_entry(input) |
| 74 | skills.hide_chat_skill(context.get_agent(), skill_entry) |
| 75 | save_tmp_chat(context) |
| 76 | return self._build_state(context_id=context.id) |
| 77 | |
| 78 | def _show(self, input: dict, *, context_id: str) -> dict[str, Any]: |
| 79 | context = self._require_context(context_id) |
| 80 | skill_entry = self._require_skill_entry(input) |
| 81 | skills.show_chat_skill(context.get_agent(), skill_entry) |
| 82 | save_tmp_chat(context) |
| 83 | return self._build_state(context_id=context.id) |
| 84 | |
| 85 | def _clear(self, *, context_id: str) -> dict[str, Any]: |
| 86 | return { |
| 87 | "ok": False, |
| 88 | "error": "Loaded skills are kept in chat history and cannot be removed.", |
| 89 | } |
| 90 | |
| 91 | def _build_state( |
| 92 | self, |
| 93 | *, |
| 94 | context_id: str = "", |
| 95 | project_name: str = "", |
| 96 | ) -> dict[str, Any]: |
| 97 | context = AgentContext.get(context_id) if context_id else None |
| 98 | agent = context.get_agent() if context else None |
| 99 | |
| 100 | if context and not project_name: |
| 101 | project_name = projects.get_context_project_name(context) or "" |
| 102 | |
| 103 | catalog = skills.list_skill_catalog(project_name=project_name, agent=agent) |
| 104 | catalog_by_key = {self._entry_key(skill): skill for skill in catalog} |
| 105 | catalog_by_name = { |
| 106 | str(skill.get("name") or "").strip().lower(): skill for skill in catalog |
| 107 | } |
| 108 | |
| 109 | loaded_entries = skills.get_loaded_skill_entries(agent) |
| 110 | scope_entries = skills.get_scope_active_skills(agent) |
| 111 | scope_hidden_entries = skills.get_scope_hidden_skills(agent) |
| 112 | chat_entries = skills.get_chat_active_skills(context) |
| 113 | disabled_entries = skills.get_chat_disabled_skills(context) |
| 114 | visible_entries = skills.get_chat_visible_skills(context) |
| 115 | hidden_entries = skills.get_hidden_skills(agent) |
| 116 | |
| 117 | return { |
| 118 | "ok": True, |
| 119 | "context_available": bool(context), |
| 120 | "context_id": context.id if context else "", |
| 121 | "project_name": project_name, |
| 122 | "skills": catalog, |
| 123 | "max_active_skills": skills.get_max_active_skills( |
| 124 | agent=agent, |
| 125 | project_name=project_name, |
| 126 | ), |
| 127 | "active_skills": [ |
| 128 | self._serialize_entry( |
| 129 | entry, |
| 130 | catalog_by_key, |
| 131 | catalog_by_name, |
| 132 | state_source="Loaded in chat history", |
| 133 | ) |
| 134 | for entry in loaded_entries |
| 135 | if (key := self._entry_key(entry)) |
| 136 | ], |
| 137 | "scope_skills": [ |
| 138 | self._serialize_entry( |
| 139 | entry, |
| 140 | catalog_by_key, |
| 141 | catalog_by_name, |
| 142 | state_source="Scope default", |
| 143 | ) |
| 144 | for entry in scope_entries |
| 145 | ], |
| 146 | "chat_skills": [ |
| 147 | self._serialize_entry( |
| 148 | entry, |
| 149 | catalog_by_key, |
| 150 | catalog_by_name, |
| 151 | state_source="Chat", |
| 152 | ) |
| 153 | for entry in chat_entries |
| 154 | ], |
| 155 | "disabled_skills": [ |
| 156 | self._serialize_entry( |
| 157 | entry, |
| 158 | catalog_by_key, |
| 159 | catalog_by_name, |
| 160 | state_source="Hidden in chat", |
| 161 | ) |
| 162 | for entry in disabled_entries |
| 163 | ], |
| 164 | "hidden_skills": [ |
| 165 | self._serialize_entry( |
| 166 | entry, |
| 167 | catalog_by_key, |
| 168 | catalog_by_name, |
| 169 | state_source=( |
| 170 | "Hidden default" |
| 171 | if self._entry_matches_any(entry, scope_hidden_entries) |
| 172 | else "Hidden in chat" |
| 173 | ), |
| 174 | ) |
| 175 | for entry in hidden_entries |
| 176 | ], |
| 177 | "scope_hidden_skills": [ |
| 178 | self._serialize_entry( |
| 179 | entry, |
| 180 | catalog_by_key, |
| 181 | catalog_by_name, |
| 182 | state_source="Hidden default", |
| 183 | ) |
| 184 | for entry in scope_hidden_entries |
| 185 | ], |
| 186 | "visible_skills": [ |
| 187 | self._serialize_entry( |
| 188 | entry, |
| 189 | catalog_by_key, |
| 190 | catalog_by_name, |
| 191 | state_source="Visible in chat", |
| 192 | ) |
| 193 | for entry in visible_entries |
| 194 | ], |
| 195 | } |
| 196 | |
| 197 | def _get_doc( |
| 198 | self, |
| 199 | input: dict, |
| 200 | *, |
| 201 | context_id: str = "", |
| 202 | project_name: str = "", |
| 203 | ) -> dict[str, Any]: |
| 204 | context = AgentContext.get(context_id) if context_id else None |
| 205 | agent = context.get_agent() if context else None |
| 206 | |
| 207 | if context and not project_name: |
| 208 | project_name = projects.get_context_project_name(context) or "" |
| 209 | |
| 210 | skill_entry = self._require_skill_entry(input) |
| 211 | requested_key = self._entry_key(skill_entry) |
| 212 | catalog = skills.list_skill_catalog(project_name=project_name, agent=agent) |
| 213 | skill = next( |
| 214 | (item for item in catalog if self._entry_key(item) == requested_key), |
| 215 | None, |
| 216 | ) |
| 217 | if not skill and skill_entry.get("name"): |
| 218 | requested_name = str(skill_entry.get("name") or "").strip().lower() |
| 219 | skill = next( |
| 220 | (item for item in catalog if str(item.get("name") or "").strip().lower() == requested_name), |
| 221 | None, |
| 222 | ) |
| 223 | |
| 224 | if not skill: |
| 225 | raise ValueError("Skill not found in the current list") |
| 226 | |
| 227 | skill_path = str(skill.get("path") or "").strip() |
| 228 | skill_md_path = Path(files.fix_dev_path(skill_path)) / "SKILL.md" |
| 229 | if not skill_md_path.is_file(): |
| 230 | raise FileNotFoundError("SKILL.md not found") |
| 231 | |
| 232 | return { |
| 233 | "ok": True, |
| 234 | "filename": f"{skill.get('name') or skill_md_path.parent.name} / SKILL.md", |
| 235 | "content": skill_md_path.read_text(encoding="utf-8", errors="replace"), |
| 236 | } |
| 237 | |
| 238 | def _require_context(self, context_id: str) -> AgentContext: |
| 239 | if not context_id: |
| 240 | raise ValueError("context_id is required") |
| 241 | |
| 242 | context = AgentContext.get(context_id) |
| 243 | if not context: |
| 244 | raise ValueError("Context not found") |
| 245 | return context |
| 246 | |
| 247 | def _require_skill_entry(self, input: dict) -> dict[str, str]: |
| 248 | entries = skills.normalize_active_skills([input.get("skill")]) |
| 249 | if not entries: |
| 250 | raise ValueError("skill is required") |
| 251 | return entries[0] |
| 252 | |
| 253 | def _entry_key(self, entry: dict[str, Any]) -> str: |
| 254 | return str(entry.get("path") or entry.get("name") or "").strip().lower() |
| 255 | |
| 256 | def _visible_skill_loaded(self, agent: Any, skill_name: str) -> bool: |
| 257 | output = getattr(getattr(agent, "history", None), "output", None) |
| 258 | if not callable(output): |
| 259 | return False |
| 260 | return any( |
| 261 | skills.skill_instruction_name(message) == skill_name |
| 262 | for message in output() |
| 263 | ) |
| 264 | |
| 265 | def _resolve_catalog_skill( |
| 266 | self, |
| 267 | entry: dict[str, Any], |
| 268 | *, |
| 269 | context: AgentContext, |
| 270 | ) -> dict[str, Any]: |
| 271 | agent = context.get_agent() |
| 272 | project_name = projects.get_context_project_name(context) or "" |
| 273 | catalog = skills.list_skill_catalog(project_name=project_name, agent=agent) |
| 274 | return next( |
| 275 | (item for item in catalog if self._entry_matches_any(entry, [item])), |
| 276 | entry, |
| 277 | ) |
| 278 | |
| 279 | def _merge_entries( |
| 280 | self, |
| 281 | *entry_groups: list[dict[str, Any]], |
| 282 | ) -> list[dict[str, Any]]: |
| 283 | merged: list[dict[str, Any]] = [] |
| 284 | seen: set[str] = set() |
| 285 | |
| 286 | for entries in entry_groups: |
| 287 | for entry in entries: |
| 288 | key = self._entry_key(entry) |
| 289 | if not key or key in seen: |
| 290 | continue |
| 291 | seen.add(key) |
| 292 | merged.append(entry) |
| 293 | |
| 294 | return merged |
| 295 | |
| 296 | def _filter_hidden_entries( |
| 297 | self, |
| 298 | entries: list[dict[str, Any]], |
| 299 | hidden_entries: list[dict[str, Any]], |
| 300 | ) -> list[dict[str, Any]]: |
| 301 | return [ |
| 302 | entry |
| 303 | for entry in entries |
| 304 | if not self._entry_matches_any(entry, hidden_entries) |
| 305 | ] |
| 306 | |
| 307 | def _entry_matches_any( |
| 308 | self, |
| 309 | entry: dict[str, Any], |
| 310 | entries: list[dict[str, Any]], |
| 311 | ) -> bool: |
| 312 | keys = { |
| 313 | str(entry.get("path") or "").strip().lower(), |
| 314 | str(entry.get("name") or "").strip().lower(), |
| 315 | } |
| 316 | keys.discard("") |
| 317 | if not keys: |
| 318 | return False |
| 319 | |
| 320 | for candidate in entries: |
| 321 | candidate_keys = { |
| 322 | str(candidate.get("path") or "").strip().lower(), |
| 323 | str(candidate.get("name") or "").strip().lower(), |
| 324 | } |
| 325 | candidate_keys.discard("") |
| 326 | if keys & candidate_keys: |
| 327 | return True |
| 328 | return False |
| 329 | |
| 330 | def _serialize_entry( |
| 331 | self, |
| 332 | entry: dict[str, Any], |
| 333 | catalog_by_key: dict[str, dict[str, Any]], |
| 334 | catalog_by_name: dict[str, dict[str, Any]], |
| 335 | *, |
| 336 | state_source: str, |
| 337 | ) -> dict[str, Any]: |
| 338 | key = self._entry_key(entry) |
| 339 | match = catalog_by_key.get(key) |
| 340 | |
| 341 | if not match: |
| 342 | name_key = str(entry.get("name") or "").strip().lower() |
| 343 | if name_key: |
| 344 | match = catalog_by_name.get(name_key) |
| 345 | |
| 346 | if match: |
| 347 | return { |
| 348 | **match, |
| 349 | "state_source": state_source, |
| 350 | "missing": False, |
| 351 | } |
| 352 | |
| 353 | path = str(entry.get("path") or "").strip() |
| 354 | fallback_name = str(entry.get("name") or "").strip() |
| 355 | if not fallback_name and path: |
| 356 | fallback_name = Path(path).name or path |
| 357 | |
| 358 | return { |
| 359 | "name": fallback_name or "(unnamed skill)", |
| 360 | "description": "", |
| 361 | "path": path, |
| 362 | "origin": "Unavailable", |
| 363 | "state_source": state_source, |
| 364 | "missing": True, |
| 365 | } |