| 1 | from __future__ import annotations |
| 2 | |
| 3 | import os |
| 4 | import re |
| 5 | from dataclasses import dataclass, field |
| 6 | from pathlib import Path |
| 7 | from typing import Any, Dict, Iterable, List, Literal, Optional, Tuple, TYPE_CHECKING, TypedDict |
| 8 | |
| 9 | from helpers import files, subagents, projects, file_tree, runtime |
| 10 | from helpers import plugins as plugin_helpers |
| 11 | |
| 12 | if TYPE_CHECKING: |
| 13 | from agent import Agent |
| 14 | |
| 15 | try: |
| 16 | import yaml # type: ignore |
| 17 | except Exception: # pragma: no cover |
| 18 | yaml = None # type: ignore |
| 19 | |
| 20 | |
| 21 | MAX_ACTIVE_SKILLS = 20 |
| 22 | ACTIVE_SKILLS_PLUGIN_NAME = "_skills" |
| 23 | AGENT_DATA_NAME_LOADED_SKILLS = "loaded_skills" |
| 24 | CONTEXT_DATA_NAME_LOADED_SKILLS = AGENT_DATA_NAME_LOADED_SKILLS |
| 25 | CONTEXT_DATA_NAME_CHAT_ACTIVE_SKILLS = "skills_chat_active" |
| 26 | CONTEXT_DATA_NAME_CHAT_DISABLED_SKILLS = "skills_chat_disabled" |
| 27 | CONTEXT_DATA_NAME_CHAT_VISIBLE_SKILLS = "skills_chat_visible" |
| 28 | _WARNED_SKILL_PARSE_PATHS: set[Path] = set() |
| 29 | |
| 30 | |
| 31 | class ActiveSkillEntry(TypedDict, total=False): |
| 32 | name: str |
| 33 | path: str |
| 34 | |
| 35 | |
| 36 | class CatalogSkill(TypedDict): |
| 37 | name: str |
| 38 | description: str |
| 39 | path: str |
| 40 | origin: str |
| 41 | hidden: bool |
| 42 | tags: list[str] |
| 43 | allowed_tools: list[str] |
| 44 | |
| 45 | |
| 46 | @dataclass(slots=True) |
| 47 | class Skill: |
| 48 | name: str |
| 49 | description: str |
| 50 | path: Path |
| 51 | skill_md_path: Path |
| 52 | version: str = "" |
| 53 | author: str = "" |
| 54 | tags: List[str] = field(default_factory=list) |
| 55 | triggers: List[str] = field(default_factory=list) |
| 56 | allowed_tools: List[str] = field(default_factory=list) |
| 57 | license: str = "" |
| 58 | compatibility: str = "" |
| 59 | metadata: Dict[str, Any] = field(default_factory=dict) |
| 60 | |
| 61 | # Optional heavy fields (only set when requested) |
| 62 | content: str = "" # body content (markdown without frontmatter) |
| 63 | raw_frontmatter: Dict[str, Any] = field(default_factory=dict) |
| 64 | |
| 65 | |
| 66 | def get_skills_base_dir() -> Path: |
| 67 | return Path(files.get_abs_path("usr", "skills")) |
| 68 | |
| 69 | |
| 70 | def get_skill_roots( |
| 71 | agent: Agent|None=None, |
| 72 | ) -> List[str]: |
| 73 | |
| 74 | if agent: |
| 75 | # skill roots available to agent |
| 76 | paths = subagents.get_paths(agent, "skills") |
| 77 | else: |
| 78 | # skill roots available globally |
| 79 | project_agents = files.find_existing_paths_by_pattern("usr/projects/*/.a0proj/agents/*/skills") # agents in projects |
| 80 | projects = files.find_existing_paths_by_pattern("usr/projects/*/.a0proj/skills") # projects |
| 81 | usr_agents = files.find_existing_paths_by_pattern("usr/agents/*/skills") # agents |
| 82 | agents = files.find_existing_paths_by_pattern("agents/*/skills") # agents |
| 83 | plugins = files.find_existing_paths_by_pattern("plugins/*/skills") # plugins |
| 84 | usr_plugins = files.find_existing_paths_by_pattern("usr/plugins/*/skills") # plugins |
| 85 | plugins_agents = files.find_existing_paths_by_pattern("plugins/*/agents/*/skills") # agents in plugins |
| 86 | usr_plugins_agents = files.find_existing_paths_by_pattern("usr/plugins/*/agents/*/skills") # agents in plugins |
| 87 | paths = [ |
| 88 | files.get_abs_path("skills"), |
| 89 | files.get_abs_path("usr/skills"), |
| 90 | *project_agents, |
| 91 | *projects, |
| 92 | *usr_agents, |
| 93 | *agents, |
| 94 | *plugins, |
| 95 | *usr_plugins, |
| 96 | *plugins_agents, |
| 97 | *usr_plugins_agents, |
| 98 | ] |
| 99 | return paths |
| 100 | |
| 101 | |
| 102 | def _is_hidden_path(path: Path) -> bool: |
| 103 | return any(part.startswith(".") for part in path.parts) |
| 104 | |
| 105 | |
| 106 | def discover_skill_md_files(root: Path) -> List[Path]: |
| 107 | """ |
| 108 | Recursively discover SKILL.md files under a root directory. |
| 109 | Hidden folders/files are ignored. |
| 110 | """ |
| 111 | if not root.exists(): |
| 112 | return [] |
| 113 | |
| 114 | results: List[Path] = [] |
| 115 | for p in root.rglob("SKILL.md"): |
| 116 | try: |
| 117 | if not p.is_file(): |
| 118 | continue |
| 119 | if _is_hidden_path(p.relative_to(root)): |
| 120 | continue |
| 121 | results.append(p) |
| 122 | except Exception: |
| 123 | # If relative_to fails (weird symlink), fall back to conservative checks |
| 124 | if p.is_file() and ".git" not in str(p): |
| 125 | results.append(p) |
| 126 | results.sort(key=lambda x: str(x)) |
| 127 | return results |
| 128 | |
| 129 | |
| 130 | def _coerce_list(value: Any) -> List[str]: |
| 131 | if value is None: |
| 132 | return [] |
| 133 | if isinstance(value, list): |
| 134 | return [str(v).strip() for v in value if str(v).strip()] |
| 135 | if isinstance(value, tuple): |
| 136 | return [str(v).strip() for v in list(value) if str(v).strip()] |
| 137 | if isinstance(value, str): |
| 138 | # Support comma-separated or space-delimited strings |
| 139 | if "," in value: |
| 140 | parts = [p.strip() for p in value.split(",")] |
| 141 | else: |
| 142 | parts = [p.strip() for p in re.split(r"\s+", value)] |
| 143 | return [p for p in parts if p] |
| 144 | return [str(value).strip()] if str(value).strip() else [] |
| 145 | |
| 146 | |
| 147 | def _normalize_name(name: str) -> str: |
| 148 | return re.sub(r"\s+", "-", (name or "").strip().lower()) |
| 149 | |
| 150 | |
| 151 | def _read_text(path: Path) -> str: |
| 152 | return path.read_text(encoding="utf-8", errors="replace") |
| 153 | |
| 154 | |
| 155 | def split_frontmatter(markdown: str) -> Tuple[Dict[str, Any], str, List[str]]: |
| 156 | """ |
| 157 | Splits a SKILL.md into (frontmatter_dict, body_text, errors). |
| 158 | Enforces YAML frontmatter at the top for spec compatibility. |
| 159 | """ |
| 160 | errors: List[str] = [] |
| 161 | text = markdown or "" |
| 162 | lines = text.splitlines() |
| 163 | |
| 164 | # Require frontmatter fence at the start (allow leading whitespace/newlines). |
| 165 | start_idx = None |
| 166 | for i, line in enumerate(lines): |
| 167 | if line.strip() == "---": |
| 168 | start_idx = i |
| 169 | break |
| 170 | if line.strip(): # non-empty before fence => invalid |
| 171 | errors.append("Frontmatter must start at the top of the file") |
| 172 | return {}, text.strip(), errors |
| 173 | |
| 174 | if start_idx is None: |
| 175 | errors.append("Missing YAML frontmatter") |
| 176 | return {}, text.strip(), errors |
| 177 | |
| 178 | end_idx = None |
| 179 | for j in range(start_idx + 1, len(lines)): |
| 180 | if lines[j].strip() == "---": |
| 181 | end_idx = j |
| 182 | break |
| 183 | |
| 184 | if end_idx is None: |
| 185 | errors.append("Unterminated YAML frontmatter") |
| 186 | return {}, text.strip(), errors |
| 187 | |
| 188 | fm_text = "\n".join(lines[start_idx + 1 : end_idx]).strip() |
| 189 | body = "\n".join(lines[end_idx + 1 :]).strip() |
| 190 | fm, fm_errors = parse_frontmatter(fm_text) |
| 191 | errors.extend(fm_errors) |
| 192 | return fm, body, errors |
| 193 | |
| 194 | |
| 195 | def _parse_frontmatter_fallback(frontmatter_text: str) -> Dict[str, Any]: |
| 196 | # Minimal YAML subset: key: value, lists with "- item" |
| 197 | data: Dict[str, Any] = {} |
| 198 | current_key: Optional[str] = None |
| 199 | for raw in frontmatter_text.splitlines(): |
| 200 | line = raw.rstrip() |
| 201 | if not line.strip() or line.strip().startswith("#"): |
| 202 | continue |
| 203 | |
| 204 | m = re.match(r"^([A-Za-z0-9_.-]+)\s*:\s*(.*)$", line) |
| 205 | if m: |
| 206 | key = m.group(1) |
| 207 | val = m.group(2).strip() |
| 208 | current_key = key |
| 209 | if val == "": |
| 210 | data[key] = [] |
| 211 | else: |
| 212 | if (val.startswith('"') and val.endswith('"')) or ( |
| 213 | val.startswith("'") and val.endswith("'") |
| 214 | ): |
| 215 | val = val[1:-1] |
| 216 | data[key] = val |
| 217 | continue |
| 218 | |
| 219 | m_list = re.match(r"^\s*-\s*(.*)$", line) |
| 220 | if m_list and current_key: |
| 221 | item = m_list.group(1).strip() |
| 222 | if (item.startswith('"') and item.endswith('"')) or ( |
| 223 | item.startswith("'") and item.endswith("'") |
| 224 | ): |
| 225 | item = item[1:-1] |
| 226 | if not isinstance(data.get(current_key), list): |
| 227 | data[current_key] = [] |
| 228 | data[current_key].append(item) |
| 229 | continue |
| 230 | return data |
| 231 | |
| 232 | |
| 233 | def parse_frontmatter(frontmatter_text: str) -> Tuple[Dict[str, Any], List[str]]: |
| 234 | """ |
| 235 | Parse YAML frontmatter with PyYAML when available, |
| 236 | falling back to a minimal subset parser. |
| 237 | """ |
| 238 | errors: List[str] = [] |
| 239 | if not frontmatter_text.strip(): |
| 240 | return {}, errors |
| 241 | |
| 242 | if yaml is not None: |
| 243 | try: |
| 244 | parsed = yaml.safe_load(frontmatter_text) # type: ignore[attr-defined] |
| 245 | except Exception as exc: |
| 246 | errors.append(f"Invalid YAML frontmatter: {exc}") |
| 247 | return {}, errors |
| 248 | if parsed is not None: |
| 249 | if not isinstance(parsed, dict): |
| 250 | errors.append("Frontmatter must be a mapping") |
| 251 | return {}, errors |
| 252 | return parsed, errors |
| 253 | |
| 254 | parsed = _parse_frontmatter_fallback(frontmatter_text) |
| 255 | if not parsed: |
| 256 | errors.append("Invalid YAML frontmatter") |
| 257 | return parsed, errors |
| 258 | |
| 259 | |
| 260 | def _emit_skill_scan_warning(message: str) -> None: |
| 261 | try: |
| 262 | from helpers.print_style import PrintStyle |
| 263 | |
| 264 | PrintStyle.warning(message) |
| 265 | except Exception: |
| 266 | print(f"Warning: {message}") |
| 267 | |
| 268 | |
| 269 | def _frontmatter_error_line(lines: List[str], error: str) -> int | None: |
| 270 | if not lines: |
| 271 | return 1 |
| 272 | |
| 273 | if error.startswith("Frontmatter must start"): |
| 274 | for index, line in enumerate(lines, start=1): |
| 275 | if line.strip(): |
| 276 | return index |
| 277 | return 1 |
| 278 | if error.startswith("Missing YAML frontmatter"): |
| 279 | return 1 |
| 280 | if error.startswith("Unterminated YAML frontmatter"): |
| 281 | return max(len(lines), 1) |
| 282 | return None |
| 283 | |
| 284 | |
| 285 | def _warn_skill_skipped(skill_md_path: Path, markdown: str, errors: List[str]) -> None: |
| 286 | if not errors: |
| 287 | return |
| 288 | if skill_md_path in _WARNED_SKILL_PARSE_PATHS: |
| 289 | return |
| 290 | _WARNED_SKILL_PARSE_PATHS.add(skill_md_path) |
| 291 | |
| 292 | error = str(errors[0] or "invalid frontmatter").strip() |
| 293 | line = _frontmatter_error_line((markdown or "").splitlines(), error) |
| 294 | skill_label = skill_md_path.parent.name or str(skill_md_path) |
| 295 | location = f" at line {line}" if line is not None else "" |
| 296 | _emit_skill_scan_warning( |
| 297 | f"skill {skill_label} skipped: invalid frontmatter{location}: {error}" |
| 298 | ) |
| 299 | |
| 300 | |
| 301 | def skill_from_markdown( |
| 302 | skill_md_path: Path, |
| 303 | *, |
| 304 | include_content: bool = False, |
| 305 | validate: bool = True, |
| 306 | ) -> Optional[Skill]: |
| 307 | try: |
| 308 | text = _read_text(skill_md_path) |
| 309 | except Exception: |
| 310 | return None |
| 311 | |
| 312 | fm, body, fm_errors = split_frontmatter(text) |
| 313 | if fm_errors: |
| 314 | _warn_skill_skipped(skill_md_path, text, fm_errors) |
| 315 | return None |
| 316 | skill_dir = Path(files.normalize_a0_path(str(skill_md_path.parent))) |
| 317 | |
| 318 | name = str(fm.get("name") or fm.get("skill") or "").strip() |
| 319 | description = str( |
| 320 | fm.get("description") or fm.get("when_to_use") or fm.get("summary") or "" |
| 321 | ).strip() |
| 322 | |
| 323 | # Cross-platform aliases: |
| 324 | # - Claude Code leans on description (triggers may be embedded there) |
| 325 | # - Some repos use triggers/trigger_patterns |
| 326 | triggers = _coerce_list( |
| 327 | fm.get("triggers") |
| 328 | or fm.get("trigger_patterns") |
| 329 | or fm.get("trigger") |
| 330 | or fm.get("activation") |
| 331 | ) |
| 332 | |
| 333 | tags = _coerce_list(fm.get("tags") or fm.get("tag")) |
| 334 | allowed_tools = _coerce_list( |
| 335 | fm.get("allowed-tools") or fm.get("allowed_tools") or fm.get("tools") |
| 336 | ) |
| 337 | |
| 338 | version = str(fm.get("version") or "").strip() |
| 339 | author = str(fm.get("author") or "").strip() |
| 340 | license_ = str(fm.get("license") or "").strip() |
| 341 | compatibility = str(fm.get("compatibility") or "").strip() |
| 342 | |
| 343 | meta = fm.get("metadata") |
| 344 | if not isinstance(meta, dict): |
| 345 | meta = {} |
| 346 | |
| 347 | skill = Skill( |
| 348 | name=name, |
| 349 | description=description, |
| 350 | path=skill_dir, |
| 351 | skill_md_path=skill_md_path, |
| 352 | version=version, |
| 353 | author=author, |
| 354 | tags=tags, |
| 355 | triggers=triggers, |
| 356 | allowed_tools=allowed_tools, |
| 357 | license=license_, |
| 358 | metadata=dict(meta), |
| 359 | compatibility=compatibility, |
| 360 | raw_frontmatter=fm if include_content else {}, |
| 361 | content=body if include_content else "", |
| 362 | ) |
| 363 | if validate: |
| 364 | issues = validate_skill(skill) |
| 365 | if issues: |
| 366 | return None |
| 367 | return skill |
| 368 | |
| 369 | |
| 370 | def list_skills( |
| 371 | agent:Agent|None=None, |
| 372 | include_content: bool = False, |
| 373 | include_hidden: bool = False, |
| 374 | ) -> List[Skill]: |
| 375 | """List skills, optionally filtered by agent scope.""" |
| 376 | skills: List[Skill] = [] |
| 377 | |
| 378 | roots = get_skill_roots(agent) |
| 379 | |
| 380 | for root in roots: |
| 381 | for skill_md in discover_skill_md_files(Path(root)): |
| 382 | s = skill_from_markdown(skill_md, include_content=include_content) |
| 383 | if s: |
| 384 | skills.append(s) |
| 385 | |
| 386 | # no deduplication for global skills |
| 387 | if not agent: |
| 388 | return skills |
| 389 | |
| 390 | # Dedupe by normalized name, preserving root_order priority (earlier wins) |
| 391 | by_name: Dict[str, Skill] = {} |
| 392 | for s in skills: |
| 393 | key = _normalize_name(s.name) or _normalize_name(s.path.name) |
| 394 | if key and key not in by_name: |
| 395 | by_name[key] = s |
| 396 | |
| 397 | result = list(by_name.values()) |
| 398 | if include_hidden: |
| 399 | return result |
| 400 | return _filter_hidden_skills(agent, result) |
| 401 | |
| 402 | |
| 403 | def delete_skill( |
| 404 | skill_path: str, |
| 405 | ) -> None: |
| 406 | """Delete a skill directory.""" |
| 407 | |
| 408 | skill_path = files.get_abs_path(skill_path) |
| 409 | if runtime.is_development(): |
| 410 | skill_path = files.fix_dev_path(skill_path) |
| 411 | |
| 412 | normalized_path = files.normalize_a0_path(skill_path) |
| 413 | if "/plugins/" in normalized_path and "/usr/plugins/" not in normalized_path: |
| 414 | raise PermissionError("Built-in plugin skills cannot be deleted") |
| 415 | |
| 416 | allowed_roots = get_skill_roots() |
| 417 | for root in allowed_roots: |
| 418 | if files.is_in_dir(skill_path, root): |
| 419 | break |
| 420 | else: |
| 421 | raise ValueError("Skill root not in current scope") |
| 422 | |
| 423 | |
| 424 | if not os.path.isdir(skill_path): |
| 425 | raise FileNotFoundError("Skill directory not found") |
| 426 | |
| 427 | # delete directory |
| 428 | files.delete_dir(skill_path) |
| 429 | |
| 430 | |
| 431 | def find_skill( |
| 432 | skill_name: str, |
| 433 | agent:Agent|None=None, |
| 434 | include_content: bool = False, |
| 435 | include_hidden: bool = False, |
| 436 | validate: bool = True, |
| 437 | ) -> Optional[Skill]: |
| 438 | target = _normalize_name(skill_name) |
| 439 | if not target: |
| 440 | return None |
| 441 | |
| 442 | roots = get_skill_roots(agent) |
| 443 | |
| 444 | for root in roots: |
| 445 | for skill_md in discover_skill_md_files(Path(root)): |
| 446 | s = skill_from_markdown( |
| 447 | skill_md, |
| 448 | include_content=include_content, |
| 449 | validate=validate, |
| 450 | ) |
| 451 | if not s: |
| 452 | continue |
| 453 | if _normalize_name(s.name) == target or _normalize_name(s.path.name) == target: |
| 454 | if not include_hidden and _skill_is_hidden_for_agent(agent, s): |
| 455 | continue |
| 456 | return s |
| 457 | return None |
| 458 | |
| 459 | def load_skill_for_agent( |
| 460 | skill_name: str, |
| 461 | agent: Agent | None = None, |
| 462 | ) -> str: |
| 463 | """Load skill and format it as a complete string for agent context.""" |
| 464 | skill = find_skill(skill_name, agent=agent, include_content=True) |
| 465 | if not skill: |
| 466 | return f"Error: skill '{skill_name}' not found" |
| 467 | |
| 468 | # Get runtime path |
| 469 | runtime_path = str(skill.path) |
| 470 | if runtime.is_development(): |
| 471 | runtime_path = files.normalize_a0_path(str(skill.path)) |
| 472 | |
| 473 | lines = [f"Skill: {skill.name}", f"Path: {runtime_path}"] |
| 474 | |
| 475 | # Metadata |
| 476 | metadata = [ |
| 477 | ("Version", skill.version), |
| 478 | ("Author", skill.author), |
| 479 | ("License", skill.license), |
| 480 | ("Compatibility", skill.compatibility), |
| 481 | ("Tags", ", ".join(skill.tags) if skill.tags else None), |
| 482 | ("Allowed tools", ", ".join(skill.allowed_tools) if skill.allowed_tools else None), |
| 483 | ("Triggers", ", ".join(skill.triggers) if skill.triggers else None), |
| 484 | ] |
| 485 | lines.extend(f"{label}: {value}" for label, value in metadata if value) |
| 486 | |
| 487 | # Description and content |
| 488 | if skill.description: |
| 489 | lines.extend(["", "Description:", skill.description.strip()]) |
| 490 | |
| 491 | lines.extend(["", "Content (SKILL.md body):", skill.content.strip() or "(empty)"]) |
| 492 | |
| 493 | # File tree |
| 494 | files_tree = _get_skill_files(skill.path) |
| 495 | lines.append("") |
| 496 | if files_tree: |
| 497 | lines.append("Files (use skills_tool action=read_file to open):") |
| 498 | lines.append(files_tree) |
| 499 | else: |
| 500 | lines.append("No additional files found.") |
| 501 | |
| 502 | return "\n".join(lines) |
| 503 | |
| 504 | |
| 505 | def skill_instruction_name(message: Any) -> str: |
| 506 | match message: |
| 507 | case { |
| 508 | "content": { |
| 509 | "skill_instructions": { |
| 510 | "content_included": included, |
| 511 | "name": name, |
| 512 | } |
| 513 | } |
| 514 | } if included: |
| 515 | return str(name or "").strip() |
| 516 | return "" |
| 517 | |
| 518 | |
| 519 | def _get_skill_files(skill_dir: Path) -> str: |
| 520 | """Get file tree for skill directory.""" |
| 521 | if not skill_dir.exists(): |
| 522 | return "" |
| 523 | |
| 524 | tree = str( |
| 525 | file_tree.file_tree( |
| 526 | str(skill_dir), |
| 527 | max_depth=10, |
| 528 | folders_first=True, |
| 529 | max_files=100, |
| 530 | max_folders=100, |
| 531 | output_mode="string", |
| 532 | max_lines=300, |
| 533 | ignore=files.read_file("conf/skill.default.gitignore"), |
| 534 | ) |
| 535 | ) |
| 536 | |
| 537 | if tree and runtime.is_development(): |
| 538 | runtime_path = files.normalize_a0_path(str(skill_dir)) |
| 539 | tree = tree.replace(str(skill_dir), runtime_path) |
| 540 | |
| 541 | return str(tree) |
| 542 | |
| 543 | def search_skills( |
| 544 | query: str, |
| 545 | limit: int = 25, |
| 546 | agent: Agent|None=None, |
| 547 | include_hidden: bool = False, |
| 548 | ) -> List[Skill]: |
| 549 | q = (query or "").strip().lower() |
| 550 | if not q: |
| 551 | return [] |
| 552 | |
| 553 | raw_terms = re.findall(r"[a-z0-9][a-z0-9_-]*", q) |
| 554 | terms = [ |
| 555 | t for t in raw_terms |
| 556 | if len(t) >= 4 or any(ch.isdigit() for ch in t) |
| 557 | ] |
| 558 | long_terms = [ |
| 559 | t for t in raw_terms |
| 560 | if len(t) >= 6 or any(ch.isdigit() for ch in t) |
| 561 | ] |
| 562 | candidates = list_skills(agent, include_hidden=include_hidden) |
| 563 | |
| 564 | scored: List[Tuple[int, Skill]] = [] |
| 565 | for s in candidates: |
| 566 | name = s.name.lower() |
| 567 | desc = (s.description or "").lower() |
| 568 | tags = [t.lower() for t in s.tags] |
| 569 | triggers = [t.lower() for t in s.triggers] |
| 570 | |
| 571 | score = 0 |
| 572 | if q == name: |
| 573 | score += 10 |
| 574 | if any(q == trigger for trigger in triggers): |
| 575 | score += 9 |
| 576 | if q in name: |
| 577 | score += 6 |
| 578 | if q in desc: |
| 579 | score += 4 |
| 580 | if any(q in tag for tag in tags): |
| 581 | score += 3 |
| 582 | if any(q in trigger or trigger in q for trigger in triggers): |
| 583 | score += 8 |
| 584 | |
| 585 | for term in terms: |
| 586 | if term in name: |
| 587 | score += 3 |
| 588 | for term in long_terms: |
| 589 | if any(term in tag for tag in tags): |
| 590 | score += 1 |
| 591 | if any(term in trigger for trigger in triggers): |
| 592 | score += 4 |
| 593 | |
| 594 | if score > 0: |
| 595 | scored.append((score, s)) |
| 596 | |
| 597 | scored.sort(key=lambda pair: (-pair[0], pair[1].name)) |
| 598 | return [s for _score, s in scored[:limit]] |
| 599 | |
| 600 | |
| 601 | _NAME_RE = re.compile(r"^[a-z0-9-]+$") |
| 602 | |
| 603 | |
| 604 | def validate_skill(skill: Skill) -> List[str]: |
| 605 | issues: List[str] = [] |
| 606 | name = (skill.name or "").strip() |
| 607 | desc = (skill.description or "").strip() |
| 608 | |
| 609 | if not name: |
| 610 | issues.append("Missing required field: name") |
| 611 | else: |
| 612 | if not (1 <= len(name) <= 64): |
| 613 | issues.append("name must be 1-64 characters") |
| 614 | if not _NAME_RE.match(name): |
| 615 | issues.append("name must use lowercase letters, numbers, and hyphens only") |
| 616 | if name.startswith("-") or name.endswith("-"): |
| 617 | issues.append("name must not start or end with a hyphen") |
| 618 | if "--" in name: |
| 619 | issues.append("name must not contain consecutive hyphens") |
| 620 | # if skill.path and _normalize_name(skill.path.name) != _normalize_name(name): |
| 621 | # issues.append("name should match the parent directory name") |
| 622 | |
| 623 | if not desc: |
| 624 | issues.append("Missing required field: description") |
| 625 | elif len(desc) > 1024: |
| 626 | issues.append("description must be <= 1024 characters") |
| 627 | |
| 628 | if skill.compatibility and len(skill.compatibility) > 500: |
| 629 | issues.append("compatibility must be <= 500 characters") |
| 630 | |
| 631 | return issues |
| 632 | |
| 633 | |
| 634 | def validate_skill_md(skill_md_path: Path) -> List[str]: |
| 635 | try: |
| 636 | text = _read_text(skill_md_path) |
| 637 | except Exception: |
| 638 | return ["Unable to read SKILL.md"] |
| 639 | |
| 640 | _fm, _body, fm_errors = split_frontmatter(text) |
| 641 | if fm_errors: |
| 642 | return fm_errors |
| 643 | |
| 644 | skill = skill_from_markdown( |
| 645 | skill_md_path, include_content=False, validate=False |
| 646 | ) |
| 647 | if not skill: |
| 648 | return ["Unable to parse SKILL.md frontmatter"] |
| 649 | return validate_skill(skill) |
| 650 | |
| 651 | |
| 652 | def _normalize_max_active_skills(value: Any) -> int: |
| 653 | if isinstance(value, bool): |
| 654 | return MAX_ACTIVE_SKILLS |
| 655 | |
| 656 | try: |
| 657 | normalized = int(value) |
| 658 | except (TypeError, ValueError): |
| 659 | return MAX_ACTIVE_SKILLS |
| 660 | |
| 661 | return normalized if normalized >= 1 else MAX_ACTIVE_SKILLS |
| 662 | |
| 663 | |
| 664 | def get_max_active_skills( |
| 665 | agent: Agent | None = None, |
| 666 | project_name: str | None = None, |
| 667 | ) -> int: |
| 668 | if agent is None and project_name is None: |
| 669 | return MAX_ACTIVE_SKILLS |
| 670 | |
| 671 | config = ( |
| 672 | plugin_helpers.get_plugin_config( |
| 673 | ACTIVE_SKILLS_PLUGIN_NAME, |
| 674 | agent=agent, |
| 675 | project_name=project_name or "", |
| 676 | agent_profile="", |
| 677 | ) |
| 678 | or {} |
| 679 | ) |
| 680 | return _normalize_max_active_skills(config.get("max_active_skills")) |
| 681 | |
| 682 | |
| 683 | def normalize_skills_config(config: dict[str, Any] | None) -> dict[str, Any]: |
| 684 | normalized = dict(config or {}) |
| 685 | max_active_skills = _normalize_max_active_skills( |
| 686 | normalized.get("max_active_skills") |
| 687 | ) |
| 688 | normalized["max_active_skills"] = max_active_skills |
| 689 | normalized["active_skills"] = normalize_active_skills( |
| 690 | normalized.get("active_skills"), |
| 691 | limit=max_active_skills, |
| 692 | ) |
| 693 | normalized["hidden_skills"] = normalize_hidden_skills( |
| 694 | normalized.get("hidden_skills") |
| 695 | ) |
| 696 | if "visibility_policy" in normalized: |
| 697 | normalized["visibility_policy"] = normalize_visibility_policy( |
| 698 | normalized.get("visibility_policy") |
| 699 | ) |
| 700 | return normalized |
| 701 | |
| 702 | |
| 703 | def normalize_visibility_policy(raw: Any) -> dict[str, Any]: |
| 704 | policy = dict(raw) if isinstance(raw, dict) else {} |
| 705 | mode = str(policy.get("mode") or "inherit").strip().lower() |
| 706 | default = str(policy.get("default") or "allow").strip().lower() |
| 707 | policy["mode"] = "custom" if mode == "custom" else "inherit" |
| 708 | policy["default"] = "block" if default == "block" else "allow" |
| 709 | for key in ("allowed", "blocked"): |
| 710 | policy[key] = [ |
| 711 | str(entry.get("name") or entry.get("path") or "") |
| 712 | for entry in normalize_hidden_skills(policy.get(key)) |
| 713 | ] |
| 714 | return policy |
| 715 | |
| 716 | |
| 717 | def get_visibility_policy(agent: Agent | None) -> dict[str, Any]: |
| 718 | if not agent: |
| 719 | return normalize_visibility_policy(None) |
| 720 | config = plugin_helpers.get_plugin_config( |
| 721 | ACTIVE_SKILLS_PLUGIN_NAME, |
| 722 | agent=agent, |
| 723 | ) or {} |
| 724 | return normalize_visibility_policy(config.get("visibility_policy")) |
| 725 | |
| 726 | |
| 727 | def is_skill_allowed( |
| 728 | policy: dict[str, Any], |
| 729 | skill_or_entry: Skill | ActiveSkillEntry | str, |
| 730 | ) -> bool: |
| 731 | if policy["mode"] != "custom": |
| 732 | return True |
| 733 | |
| 734 | aliases = _skill_visibility_aliases(skill_or_entry) |
| 735 | if any( |
| 736 | aliases & _skill_visibility_aliases(value) |
| 737 | for value in policy["blocked"] |
| 738 | ): |
| 739 | return False |
| 740 | if any( |
| 741 | aliases & _skill_visibility_aliases(value) |
| 742 | for value in policy["allowed"] |
| 743 | ): |
| 744 | return True |
| 745 | return policy["default"] == "allow" |
| 746 | |
| 747 | |
| 748 | def ensure_skill_visible(agent: Agent, entry: ActiveSkillEntry | str) -> None: |
| 749 | if is_skill_allowed(get_visibility_policy(agent), entry): |
| 750 | return |
| 751 | name = ( |
| 752 | str(entry.get("name") or entry.get("path") or "").strip() |
| 753 | if isinstance(entry, dict) |
| 754 | else str(entry or "").strip() |
| 755 | ) |
| 756 | profile = str(getattr(getattr(agent, "config", None), "profile", "") or "default") |
| 757 | raise ValueError(f'Skill "{name}" is blocked for agent profile "{profile}".') |
| 758 | |
| 759 | |
| 760 | def normalize_active_skills( |
| 761 | raw: Any, |
| 762 | *, |
| 763 | limit: int | None = None, |
| 764 | ) -> list[ActiveSkillEntry]: |
| 765 | return normalize_skill_entries( |
| 766 | raw, |
| 767 | limit=get_max_active_skills() if limit is None else limit, |
| 768 | ) |
| 769 | |
| 770 | |
| 771 | def normalize_hidden_skills(raw: Any) -> list[ActiveSkillEntry]: |
| 772 | return normalize_skill_entries(raw, limit=None) |
| 773 | |
| 774 | |
| 775 | def normalize_skill_entries( |
| 776 | raw: Any, |
| 777 | *, |
| 778 | limit: int | None = None, |
| 779 | ) -> list[ActiveSkillEntry]: |
| 780 | if not isinstance(raw, list): |
| 781 | return [] |
| 782 | |
| 783 | normalized: list[ActiveSkillEntry] = [] |
| 784 | seen: set[str] = set() |
| 785 | |
| 786 | for item in raw: |
| 787 | entry = _normalize_active_skill_entry(item) |
| 788 | if not entry: |
| 789 | continue |
| 790 | |
| 791 | key = _entry_key(entry) |
| 792 | if not key or key in seen: |
| 793 | continue |
| 794 | |
| 795 | seen.add(key) |
| 796 | normalized.append(entry) |
| 797 | if limit is not None and len(normalized) >= limit: |
| 798 | break |
| 799 | |
| 800 | return normalized |
| 801 | |
| 802 | |
| 803 | def list_skill_catalog( |
| 804 | project_name: str = "", |
| 805 | agent: Agent | None = None, |
| 806 | ) -> list[CatalogSkill]: |
| 807 | if not project_name: |
| 808 | project_name = _get_agent_project_name(agent) |
| 809 | |
| 810 | catalog: list[CatalogSkill] = [] |
| 811 | seen_paths: set[str] = set() |
| 812 | hidden_entries = get_hidden_skills(agent) if agent else [] |
| 813 | visibility_policy = get_visibility_policy(agent) |
| 814 | |
| 815 | for root in _get_catalog_roots(project_name=project_name, agent=agent): |
| 816 | root_path = Path(root) |
| 817 | for skill_md in discover_skill_md_files(root_path): |
| 818 | skill = skill_from_markdown(skill_md, include_content=False) |
| 819 | if not skill: |
| 820 | continue |
| 821 | |
| 822 | runtime_path = files.normalize_a0_path(str(skill.path)) |
| 823 | if runtime_path in seen_paths: |
| 824 | continue |
| 825 | |
| 826 | seen_paths.add(runtime_path) |
| 827 | allowed = is_skill_allowed(visibility_policy, skill) |
| 828 | catalog.append( |
| 829 | { |
| 830 | "name": skill.name or skill.path.name, |
| 831 | "description": skill.description or "", |
| 832 | "path": runtime_path, |
| 833 | "origin": _get_skill_origin( |
| 834 | runtime_path, |
| 835 | project_name=project_name, |
| 836 | ), |
| 837 | "hidden": _skill_matches_entries( |
| 838 | skill, hidden_entries |
| 839 | ) or not allowed, |
| 840 | "tags": list(skill.tags), |
| 841 | "allowed_tools": list(skill.allowed_tools), |
| 842 | } |
| 843 | ) |
| 844 | |
| 845 | catalog.sort(key=lambda item: (item["name"].lower(), item["path"])) |
| 846 | return catalog |
| 847 | |
| 848 | |
| 849 | def get_scope_active_skills(agent: Agent | None) -> list[ActiveSkillEntry]: |
| 850 | if not agent: |
| 851 | return [] |
| 852 | |
| 853 | project_name = _get_agent_project_name(agent) |
| 854 | config = ( |
| 855 | plugin_helpers.get_plugin_config( |
| 856 | ACTIVE_SKILLS_PLUGIN_NAME, |
| 857 | agent=agent, |
| 858 | project_name=project_name, |
| 859 | agent_profile="", |
| 860 | ) |
| 861 | or {} |
| 862 | ) |
| 863 | return normalize_active_skills( |
| 864 | config.get("active_skills"), |
| 865 | limit=get_max_active_skills(agent=agent, project_name=project_name), |
| 866 | ) |
| 867 | |
| 868 | |
| 869 | def get_scope_hidden_skills(agent: Agent | None) -> list[ActiveSkillEntry]: |
| 870 | if not agent: |
| 871 | return [] |
| 872 | |
| 873 | project_name = _get_agent_project_name(agent) |
| 874 | config = ( |
| 875 | plugin_helpers.get_plugin_config( |
| 876 | ACTIVE_SKILLS_PLUGIN_NAME, |
| 877 | agent=agent, |
| 878 | project_name=project_name, |
| 879 | agent_profile="", |
| 880 | ) |
| 881 | or {} |
| 882 | ) |
| 883 | return normalize_hidden_skills(config.get("hidden_skills")) |
| 884 | |
| 885 | |
| 886 | def get_chat_active_skills(context: Any | None) -> list[ActiveSkillEntry]: |
| 887 | if not context: |
| 888 | return [] |
| 889 | agent = context.get_agent() if hasattr(context, "get_agent") else None |
| 890 | return normalize_active_skills( |
| 891 | context.get_data(CONTEXT_DATA_NAME_CHAT_ACTIVE_SKILLS), |
| 892 | limit=get_max_active_skills(agent=agent), |
| 893 | ) |
| 894 | |
| 895 | |
| 896 | def get_chat_disabled_skills(context: Any | None) -> list[ActiveSkillEntry]: |
| 897 | if not context: |
| 898 | return [] |
| 899 | return normalize_hidden_skills( |
| 900 | context.get_data(CONTEXT_DATA_NAME_CHAT_DISABLED_SKILLS) |
| 901 | ) |
| 902 | |
| 903 | |
| 904 | def get_chat_visible_skills(context: Any | None) -> list[ActiveSkillEntry]: |
| 905 | if not context: |
| 906 | return [] |
| 907 | return normalize_hidden_skills( |
| 908 | context.get_data(CONTEXT_DATA_NAME_CHAT_VISIBLE_SKILLS) |
| 909 | ) |
| 910 | |
| 911 | |
| 912 | def get_hidden_skills(agent: Agent | None) -> list[ActiveSkillEntry]: |
| 913 | if not agent: |
| 914 | return [] |
| 915 | |
| 916 | context = getattr(agent, "context", None) |
| 917 | return _merge_hidden_skill_entries( |
| 918 | get_scope_hidden_skills(agent), |
| 919 | get_chat_disabled_skills(context), |
| 920 | get_chat_visible_skills(context), |
| 921 | ) |
| 922 | |
| 923 | |
| 924 | def _build_active_skills( |
| 925 | agent: Agent | None, |
| 926 | *, |
| 927 | chat_entries: list[ActiveSkillEntry] | None = None, |
| 928 | hidden_entries: list[ActiveSkillEntry] | None = None, |
| 929 | visible_entries: list[ActiveSkillEntry] | None = None, |
| 930 | limit: int | None = None, |
| 931 | ) -> list[ActiveSkillEntry]: |
| 932 | if not agent: |
| 933 | return [] |
| 934 | |
| 935 | context = getattr(agent, "context", None) |
| 936 | effective_limit = get_max_active_skills(agent=agent) if limit is None else limit |
| 937 | scope_entries = get_scope_active_skills(agent) |
| 938 | current_chat_entries = list( |
| 939 | chat_entries if chat_entries is not None else get_chat_active_skills(context) |
| 940 | ) |
| 941 | current_hidden_entries = list( |
| 942 | hidden_entries |
| 943 | if hidden_entries is not None |
| 944 | else get_chat_disabled_skills(context) |
| 945 | ) |
| 946 | current_visible_entries = list( |
| 947 | visible_entries |
| 948 | if visible_entries is not None |
| 949 | else get_chat_visible_skills(context) |
| 950 | ) |
| 951 | effective_hidden_entries = _merge_hidden_skill_entries( |
| 952 | get_scope_hidden_skills(agent), |
| 953 | current_hidden_entries, |
| 954 | current_visible_entries, |
| 955 | ) |
| 956 | merged = _merge_active_skill_entries( |
| 957 | scope_entries, |
| 958 | current_chat_entries, |
| 959 | effective_hidden_entries, |
| 960 | limit=effective_limit, |
| 961 | ) |
| 962 | visibility_policy = get_visibility_policy(agent) |
| 963 | return [ |
| 964 | entry for entry in merged if is_skill_allowed(visibility_policy, entry) |
| 965 | ] |
| 966 | |
| 967 | |
| 968 | def get_active_skills(agent: Agent | None) -> list[ActiveSkillEntry]: |
| 969 | return _build_active_skills(agent, limit=get_max_active_skills(agent=agent)) |
| 970 | |
| 971 | |
| 972 | def _normalize_loaded_skill_names(raw: Any) -> list[str]: |
| 973 | if not isinstance(raw, list): |
| 974 | return [] |
| 975 | |
| 976 | names: list[str] = [] |
| 977 | for value in raw: |
| 978 | name = str(value or "").strip() |
| 979 | if name and name not in names: |
| 980 | names.append(name) |
| 981 | return names |
| 982 | |
| 983 | |
| 984 | def get_loaded_skill_names(agent: Agent | None) -> list[str]: |
| 985 | if not agent: |
| 986 | return [] |
| 987 | |
| 988 | context = getattr(agent, "context", None) |
| 989 | if context and hasattr(context, "get_data"): |
| 990 | names = _normalize_loaded_skill_names( |
| 991 | context.get_data(CONTEXT_DATA_NAME_LOADED_SKILLS) |
| 992 | ) |
| 993 | if names: |
| 994 | data = getattr(agent, "data", None) |
| 995 | if isinstance(data, dict): |
| 996 | data.pop(AGENT_DATA_NAME_LOADED_SKILLS, None) |
| 997 | return names |
| 998 | |
| 999 | legacy_names = _normalize_loaded_skill_names( |
| 1000 | getattr(agent, "data", {}).get(AGENT_DATA_NAME_LOADED_SKILLS) |
| 1001 | ) |
| 1002 | if legacy_names: |
| 1003 | set_loaded_skill_names(agent, legacy_names) |
| 1004 | return legacy_names |
| 1005 | |
| 1006 | |
| 1007 | def set_loaded_skill_names(agent: Agent | None, skill_names: Any) -> list[str]: |
| 1008 | names = _normalize_loaded_skill_names(skill_names)[-MAX_ACTIVE_SKILLS:] |
| 1009 | if not agent: |
| 1010 | return names |
| 1011 | |
| 1012 | context = getattr(agent, "context", None) |
| 1013 | if context and hasattr(context, "set_data"): |
| 1014 | context.set_data(CONTEXT_DATA_NAME_LOADED_SKILLS, names or None) |
| 1015 | data = getattr(agent, "data", None) |
| 1016 | if isinstance(data, dict): |
| 1017 | data.pop(AGENT_DATA_NAME_LOADED_SKILLS, None) |
| 1018 | return names |
| 1019 | |
| 1020 | data = getattr(agent, "data", None) |
| 1021 | if isinstance(data, dict): |
| 1022 | data[AGENT_DATA_NAME_LOADED_SKILLS] = names |
| 1023 | return names |
| 1024 | |
| 1025 | |
| 1026 | def add_loaded_skill_name( |
| 1027 | agent: Agent | None, |
| 1028 | skill_name: str, |
| 1029 | *, |
| 1030 | limit: int | None = None, |
| 1031 | ) -> list[str]: |
| 1032 | name = str(skill_name or "").strip() |
| 1033 | if not name: |
| 1034 | return get_loaded_skill_names(agent) |
| 1035 | |
| 1036 | names = [loaded for loaded in get_loaded_skill_names(agent) if loaded != name] |
| 1037 | names.append(name) |
| 1038 | return set_loaded_skill_names(agent, names[-(limit or MAX_ACTIVE_SKILLS):]) |
| 1039 | |
| 1040 | |
| 1041 | def get_loaded_skill_entries(agent: Agent | None) -> list[ActiveSkillEntry]: |
| 1042 | return [{"name": skill_name} for skill_name in get_loaded_skill_names(agent)] |
| 1043 | |
| 1044 | |
| 1045 | def unload_agent_skill(agent: Agent | None, entry: Any) -> bool: |
| 1046 | normalized = _normalize_active_skill_entry(entry) |
| 1047 | if not agent or not normalized: |
| 1048 | return False |
| 1049 | |
| 1050 | next_loaded: list[str] = [] |
| 1051 | removed = False |
| 1052 | for skill_name in get_loaded_skill_names(agent): |
| 1053 | loaded_entry = _normalize_active_skill_entry(str(skill_name)) |
| 1054 | if loaded_entry and _entries_match(loaded_entry, normalized): |
| 1055 | removed = True |
| 1056 | continue |
| 1057 | next_loaded.append(skill_name) |
| 1058 | |
| 1059 | if removed: |
| 1060 | set_loaded_skill_names(agent, next_loaded) |
| 1061 | return removed |
| 1062 | |
| 1063 | |
| 1064 | def activate_chat_skill(agent: Agent, entry: Any) -> list[ActiveSkillEntry]: |
| 1065 | normalized = _normalize_active_skill_entry(entry) |
| 1066 | if not normalized: |
| 1067 | raise ValueError("A skill name or path is required.") |
| 1068 | ensure_skill_visible(agent, normalized) |
| 1069 | |
| 1070 | context = getattr(agent, "context", None) |
| 1071 | if not context: |
| 1072 | raise ValueError("A chat context is required.") |
| 1073 | |
| 1074 | scope_entries = get_scope_active_skills(agent) |
| 1075 | chat_entries = [ |
| 1076 | item |
| 1077 | for item in get_chat_active_skills(context) |
| 1078 | if not _entries_match(item, normalized) |
| 1079 | ] |
| 1080 | hidden_entries = [ |
| 1081 | item |
| 1082 | for item in get_chat_disabled_skills(context) |
| 1083 | if not _entries_match(item, normalized) |
| 1084 | ] |
| 1085 | visible_entries = [ |
| 1086 | item |
| 1087 | for item in get_chat_visible_skills(context) |
| 1088 | if not _entries_match(item, normalized) |
| 1089 | ] |
| 1090 | |
| 1091 | if not any(_entries_match(item, normalized) for item in scope_entries): |
| 1092 | chat_entries.append(normalized) |
| 1093 | if _entry_matches_any(normalized, get_scope_hidden_skills(agent)): |
| 1094 | visible_entries.append(normalized) |
| 1095 | |
| 1096 | merged_entries = _build_active_skills( |
| 1097 | agent, |
| 1098 | chat_entries=chat_entries, |
| 1099 | hidden_entries=hidden_entries, |
| 1100 | visible_entries=visible_entries, |
| 1101 | limit=-1, |
| 1102 | ) |
| 1103 | max_active_skills = get_max_active_skills(agent=agent) |
| 1104 | if len(merged_entries) > max_active_skills: |
| 1105 | raise ValueError( |
| 1106 | f"You can activate at most {max_active_skills} skills." |
| 1107 | ) |
| 1108 | |
| 1109 | _store_context_active_skill_entries( |
| 1110 | context, |
| 1111 | CONTEXT_DATA_NAME_CHAT_ACTIVE_SKILLS, |
| 1112 | chat_entries, |
| 1113 | ) |
| 1114 | _store_context_hidden_skill_entries( |
| 1115 | context, |
| 1116 | CONTEXT_DATA_NAME_CHAT_DISABLED_SKILLS, |
| 1117 | hidden_entries, |
| 1118 | ) |
| 1119 | _store_context_hidden_skill_entries( |
| 1120 | context, |
| 1121 | CONTEXT_DATA_NAME_CHAT_VISIBLE_SKILLS, |
| 1122 | visible_entries, |
| 1123 | ) |
| 1124 | return get_active_skills(agent) |
| 1125 | |
| 1126 | |
| 1127 | def deactivate_chat_skill(agent: Agent, entry: Any) -> list[ActiveSkillEntry]: |
| 1128 | normalized = _normalize_active_skill_entry(entry) |
| 1129 | if not normalized: |
| 1130 | raise ValueError("A skill name or path is required.") |
| 1131 | |
| 1132 | context = getattr(agent, "context", None) |
| 1133 | if not context: |
| 1134 | raise ValueError("A chat context is required.") |
| 1135 | |
| 1136 | chat_entries = [ |
| 1137 | item |
| 1138 | for item in get_chat_active_skills(context) |
| 1139 | if not _entries_match(item, normalized) |
| 1140 | ] |
| 1141 | hidden_entries = [ |
| 1142 | item |
| 1143 | for item in get_chat_disabled_skills(context) |
| 1144 | if not _entries_match(item, normalized) |
| 1145 | ] |
| 1146 | visible_entries = [ |
| 1147 | item |
| 1148 | for item in get_chat_visible_skills(context) |
| 1149 | if not _entries_match(item, normalized) |
| 1150 | ] |
| 1151 | |
| 1152 | is_scope_default = any( |
| 1153 | _entries_match(item, normalized) for item in get_scope_active_skills(agent) |
| 1154 | ) |
| 1155 | if is_scope_default: |
| 1156 | hidden_entries.append(normalized) |
| 1157 | |
| 1158 | _store_context_active_skill_entries( |
| 1159 | context, |
| 1160 | CONTEXT_DATA_NAME_CHAT_ACTIVE_SKILLS, |
| 1161 | chat_entries, |
| 1162 | ) |
| 1163 | _store_context_hidden_skill_entries( |
| 1164 | context, |
| 1165 | CONTEXT_DATA_NAME_CHAT_DISABLED_SKILLS, |
| 1166 | hidden_entries, |
| 1167 | ) |
| 1168 | _store_context_hidden_skill_entries( |
| 1169 | context, |
| 1170 | CONTEXT_DATA_NAME_CHAT_VISIBLE_SKILLS, |
| 1171 | visible_entries, |
| 1172 | ) |
| 1173 | return get_active_skills(agent) |
| 1174 | |
| 1175 | |
| 1176 | def hide_chat_skill(agent: Agent, entry: Any) -> list[ActiveSkillEntry]: |
| 1177 | normalized = _normalize_active_skill_entry(entry) |
| 1178 | if not normalized: |
| 1179 | raise ValueError("A skill name or path is required.") |
| 1180 | |
| 1181 | context = getattr(agent, "context", None) |
| 1182 | if not context: |
| 1183 | raise ValueError("A chat context is required.") |
| 1184 | |
| 1185 | chat_entries = [ |
| 1186 | item |
| 1187 | for item in get_chat_active_skills(context) |
| 1188 | if not _entries_match(item, normalized) |
| 1189 | ] |
| 1190 | hidden_entries = [ |
| 1191 | item |
| 1192 | for item in get_chat_disabled_skills(context) |
| 1193 | if not _entries_match(item, normalized) |
| 1194 | ] |
| 1195 | hidden_entries.append(normalized) |
| 1196 | visible_entries = [ |
| 1197 | item |
| 1198 | for item in get_chat_visible_skills(context) |
| 1199 | if not _entries_match(item, normalized) |
| 1200 | ] |
| 1201 | |
| 1202 | _store_context_active_skill_entries( |
| 1203 | context, |
| 1204 | CONTEXT_DATA_NAME_CHAT_ACTIVE_SKILLS, |
| 1205 | chat_entries, |
| 1206 | ) |
| 1207 | _store_context_hidden_skill_entries( |
| 1208 | context, |
| 1209 | CONTEXT_DATA_NAME_CHAT_DISABLED_SKILLS, |
| 1210 | hidden_entries, |
| 1211 | ) |
| 1212 | _store_context_hidden_skill_entries( |
| 1213 | context, |
| 1214 | CONTEXT_DATA_NAME_CHAT_VISIBLE_SKILLS, |
| 1215 | visible_entries, |
| 1216 | ) |
| 1217 | return get_hidden_skills(agent) |
| 1218 | |
| 1219 | |
| 1220 | def show_chat_skill(agent: Agent, entry: Any) -> list[ActiveSkillEntry]: |
| 1221 | normalized = _normalize_active_skill_entry(entry) |
| 1222 | if not normalized: |
| 1223 | raise ValueError("A skill name or path is required.") |
| 1224 | ensure_skill_visible(agent, normalized) |
| 1225 | |
| 1226 | context = getattr(agent, "context", None) |
| 1227 | if not context: |
| 1228 | raise ValueError("A chat context is required.") |
| 1229 | |
| 1230 | hidden_entries = [ |
| 1231 | item |
| 1232 | for item in get_chat_disabled_skills(context) |
| 1233 | if not _entries_match(item, normalized) |
| 1234 | ] |
| 1235 | visible_entries = [ |
| 1236 | item |
| 1237 | for item in get_chat_visible_skills(context) |
| 1238 | if not _entries_match(item, normalized) |
| 1239 | ] |
| 1240 | if _entry_matches_any(normalized, get_scope_hidden_skills(agent)): |
| 1241 | visible_entries.append(normalized) |
| 1242 | |
| 1243 | _store_context_hidden_skill_entries( |
| 1244 | context, |
| 1245 | CONTEXT_DATA_NAME_CHAT_DISABLED_SKILLS, |
| 1246 | hidden_entries, |
| 1247 | ) |
| 1248 | _store_context_hidden_skill_entries( |
| 1249 | context, |
| 1250 | CONTEXT_DATA_NAME_CHAT_VISIBLE_SKILLS, |
| 1251 | visible_entries, |
| 1252 | ) |
| 1253 | return get_hidden_skills(agent) |
| 1254 | |
| 1255 | |
| 1256 | def clear_chat_skill_overrides(agent: Agent) -> list[ActiveSkillEntry]: |
| 1257 | context = getattr(agent, "context", None) |
| 1258 | if not context: |
| 1259 | raise ValueError("A chat context is required.") |
| 1260 | |
| 1261 | _store_context_active_skill_entries(context, CONTEXT_DATA_NAME_CHAT_ACTIVE_SKILLS, []) |
| 1262 | _store_context_hidden_skill_entries(context, CONTEXT_DATA_NAME_CHAT_DISABLED_SKILLS, []) |
| 1263 | _store_context_hidden_skill_entries(context, CONTEXT_DATA_NAME_CHAT_VISIBLE_SKILLS, []) |
| 1264 | return get_active_skills(agent) |
| 1265 | |
| 1266 | |
| 1267 | def build_active_skills_prompt(agent: Agent | None) -> str: |
| 1268 | return "" |
| 1269 | |
| 1270 | |
| 1271 | def _format_skill_prompt(skill: Skill) -> str: |
| 1272 | lines = [ |
| 1273 | f"Skill: {skill.name or skill.path.name}", |
| 1274 | f"Path: {files.normalize_a0_path(str(skill.path))}", |
| 1275 | ] |
| 1276 | |
| 1277 | if skill.description: |
| 1278 | lines.extend(["", "Description:", skill.description.strip()]) |
| 1279 | |
| 1280 | lines.extend(["", "Instructions:", (skill.content or "").strip() or "(empty)"]) |
| 1281 | return "\n".join(lines) |
| 1282 | |
| 1283 | |
| 1284 | def _get_skill_origin(skill_path: str, project_name: str = "") -> str: |
| 1285 | abs_path = files.fix_dev_path(skill_path) |
| 1286 | |
| 1287 | if project_name: |
| 1288 | project_root = projects.get_project_meta(project_name, "skills") |
| 1289 | if files.exists(project_root) and files.is_in_dir(abs_path, project_root): |
| 1290 | return "Project" |
| 1291 | |
| 1292 | user_root = files.get_abs_path("usr", "skills") |
| 1293 | if files.exists(user_root) and files.is_in_dir(abs_path, user_root): |
| 1294 | return "User" |
| 1295 | |
| 1296 | normalized_path = files.normalize_a0_path(abs_path) |
| 1297 | if "/usr/plugins/" in normalized_path: |
| 1298 | return "Community plugin" |
| 1299 | if "/plugins/" in normalized_path: |
| 1300 | return "Built-in plugin" |
| 1301 | return "Built-in" |
| 1302 | |
| 1303 | |
| 1304 | def _normalize_active_skill_entry(item: Any) -> ActiveSkillEntry | None: |
| 1305 | if isinstance(item, str): |
| 1306 | stripped = item.strip() |
| 1307 | if not stripped: |
| 1308 | return None |
| 1309 | if "/" in stripped: |
| 1310 | return {"path": _normalize_active_skill_path(stripped)} |
| 1311 | return {"name": stripped} |
| 1312 | |
| 1313 | if not isinstance(item, dict): |
| 1314 | return None |
| 1315 | |
| 1316 | name = str(item.get("name") or "").strip() |
| 1317 | path = str(item.get("path") or "").strip() |
| 1318 | |
| 1319 | if path: |
| 1320 | path = _normalize_active_skill_path(path) |
| 1321 | if not (path or name): |
| 1322 | return None |
| 1323 | |
| 1324 | entry: ActiveSkillEntry = {} |
| 1325 | if name: |
| 1326 | entry["name"] = name |
| 1327 | if path: |
| 1328 | entry["path"] = path |
| 1329 | return entry |
| 1330 | |
| 1331 | |
| 1332 | def _normalize_active_skill_path(path: str) -> str: |
| 1333 | fixed = path.strip().replace("\\", "/") |
| 1334 | if fixed.startswith("/a0/"): |
| 1335 | return fixed.rstrip("/") |
| 1336 | if fixed.startswith("/"): |
| 1337 | return files.normalize_a0_path(fixed).rstrip("/") |
| 1338 | return files.normalize_a0_path(files.get_abs_path(fixed)).rstrip("/") |
| 1339 | |
| 1340 | |
| 1341 | def _entry_key(entry: ActiveSkillEntry) -> str: |
| 1342 | return str(entry.get("path") or entry.get("name") or "").strip().lower() |
| 1343 | |
| 1344 | |
| 1345 | def _entry_keys(entry: ActiveSkillEntry) -> set[str]: |
| 1346 | keys: set[str] = set() |
| 1347 | for value in (entry.get("path"), entry.get("name")): |
| 1348 | key = str(value or "").strip().lower() |
| 1349 | if key: |
| 1350 | keys.add(key) |
| 1351 | return keys |
| 1352 | |
| 1353 | |
| 1354 | def _entries_match(left: ActiveSkillEntry, right: ActiveSkillEntry) -> bool: |
| 1355 | return bool(_entry_keys(left) & _entry_keys(right)) |
| 1356 | |
| 1357 | |
| 1358 | def _entry_matches_any( |
| 1359 | entry: ActiveSkillEntry, |
| 1360 | entries: list[ActiveSkillEntry], |
| 1361 | ) -> bool: |
| 1362 | return any(_entries_match(item, entry) for item in entries) |
| 1363 | |
| 1364 | |
| 1365 | def _get_agent_project_name(agent: Agent | None) -> str: |
| 1366 | context = getattr(agent, "context", None) |
| 1367 | if not context: |
| 1368 | return "" |
| 1369 | return projects.get_context_project_name(context) or "" |
| 1370 | |
| 1371 | |
| 1372 | def _get_catalog_roots( |
| 1373 | project_name: str = "", |
| 1374 | agent: Agent | None = None, |
| 1375 | ) -> list[str]: |
| 1376 | roots: list[str] = [] |
| 1377 | seen: set[str] = set() |
| 1378 | |
| 1379 | def add(path: str) -> None: |
| 1380 | if not path: |
| 1381 | return |
| 1382 | fixed = files.fix_dev_path(path) |
| 1383 | if not files.exists(fixed) or fixed in seen: |
| 1384 | return |
| 1385 | seen.add(fixed) |
| 1386 | roots.append(fixed) |
| 1387 | |
| 1388 | if agent is not None: |
| 1389 | for path in get_skill_roots(agent): |
| 1390 | add(path) |
| 1391 | return roots |
| 1392 | |
| 1393 | if project_name: |
| 1394 | add(projects.get_project_meta(project_name, "skills")) |
| 1395 | |
| 1396 | add(files.get_abs_path("usr", "skills")) |
| 1397 | for path in plugin_helpers.get_enabled_plugin_paths(None, "skills"): |
| 1398 | add(path) |
| 1399 | add(files.get_abs_path("skills")) |
| 1400 | |
| 1401 | return roots |
| 1402 | |
| 1403 | |
| 1404 | def _merge_active_skill_entries( |
| 1405 | scope_entries: list[ActiveSkillEntry], |
| 1406 | dynamic_entries: list[ActiveSkillEntry], |
| 1407 | disabled_entries: list[ActiveSkillEntry], |
| 1408 | *, |
| 1409 | limit: int | None, |
| 1410 | ) -> list[ActiveSkillEntry]: |
| 1411 | merged: list[ActiveSkillEntry] = [] |
| 1412 | seen: set[str] = set() |
| 1413 | disabled_keys = { |
| 1414 | key for entry in disabled_entries for key in _entry_keys(entry) |
| 1415 | } |
| 1416 | |
| 1417 | for entry in [*scope_entries, *dynamic_entries]: |
| 1418 | keys = _entry_keys(entry) |
| 1419 | key = _entry_key(entry) |
| 1420 | if not key or keys & seen or keys & disabled_keys: |
| 1421 | continue |
| 1422 | |
| 1423 | seen.update(keys) |
| 1424 | merged.append(entry) |
| 1425 | if limit is not None and limit >= 0 and len(merged) >= limit: |
| 1426 | break |
| 1427 | |
| 1428 | return merged |
| 1429 | |
| 1430 | |
| 1431 | def _merge_hidden_skill_entries( |
| 1432 | scope_entries: list[ActiveSkillEntry], |
| 1433 | chat_hidden_entries: list[ActiveSkillEntry], |
| 1434 | chat_visible_entries: list[ActiveSkillEntry], |
| 1435 | ) -> list[ActiveSkillEntry]: |
| 1436 | merged: list[ActiveSkillEntry] = [] |
| 1437 | seen: set[str] = set() |
| 1438 | visible_keys = { |
| 1439 | key for entry in chat_visible_entries for key in _entry_keys(entry) |
| 1440 | } |
| 1441 | |
| 1442 | for entry in [*scope_entries, *chat_hidden_entries]: |
| 1443 | keys = _entry_keys(entry) |
| 1444 | key = _entry_key(entry) |
| 1445 | if not key or keys & seen or keys & visible_keys: |
| 1446 | continue |
| 1447 | seen.update(keys) |
| 1448 | merged.append(entry) |
| 1449 | |
| 1450 | return merged |
| 1451 | |
| 1452 | |
| 1453 | def _store_context_active_skill_entries( |
| 1454 | context: Any, |
| 1455 | key: str, |
| 1456 | entries: list[ActiveSkillEntry], |
| 1457 | ) -> None: |
| 1458 | agent = context.get_agent() if hasattr(context, "get_agent") else None |
| 1459 | normalized_entries = normalize_active_skills( |
| 1460 | entries, |
| 1461 | limit=get_max_active_skills(agent=agent), |
| 1462 | ) |
| 1463 | context.set_data(key, normalized_entries or None) |
| 1464 | |
| 1465 | |
| 1466 | def _store_context_hidden_skill_entries( |
| 1467 | context: Any, |
| 1468 | key: str, |
| 1469 | entries: list[ActiveSkillEntry], |
| 1470 | ) -> None: |
| 1471 | normalized_entries = normalize_hidden_skills(entries) |
| 1472 | context.set_data(key, normalized_entries or None) |
| 1473 | |
| 1474 | |
| 1475 | def _resolve_active_skill_entries( |
| 1476 | agent: Agent | None, |
| 1477 | entries: list[ActiveSkillEntry], |
| 1478 | ) -> list[dict[str, str]]: |
| 1479 | if not agent: |
| 1480 | return [] |
| 1481 | |
| 1482 | visible_roots = [files.fix_dev_path(root) for root in get_skill_roots(agent)] |
| 1483 | resolved: list[dict[str, str]] = [] |
| 1484 | seen_paths: set[str] = set() |
| 1485 | |
| 1486 | for entry in entries: |
| 1487 | skill = _resolve_active_skill_entry(entry, visible_roots) |
| 1488 | if not skill: |
| 1489 | continue |
| 1490 | |
| 1491 | runtime_path = files.normalize_a0_path(str(skill.path)) |
| 1492 | if runtime_path in seen_paths: |
| 1493 | continue |
| 1494 | |
| 1495 | seen_paths.add(runtime_path) |
| 1496 | resolved.append( |
| 1497 | { |
| 1498 | "name": skill.name or skill.path.name, |
| 1499 | "path": runtime_path, |
| 1500 | "content": _format_skill_prompt(skill), |
| 1501 | } |
| 1502 | ) |
| 1503 | |
| 1504 | return resolved |
| 1505 | |
| 1506 | |
| 1507 | def _resolve_active_skill_entry( |
| 1508 | entry: ActiveSkillEntry, |
| 1509 | visible_roots: list[str], |
| 1510 | ) -> Skill | None: |
| 1511 | skill_path = str(entry.get("path") or "").strip() |
| 1512 | if skill_path: |
| 1513 | skill = _load_skill_from_runtime_path(skill_path, visible_roots) |
| 1514 | if skill: |
| 1515 | return skill |
| 1516 | |
| 1517 | skill_name = str(entry.get("name") or "").strip() |
| 1518 | if not skill_name: |
| 1519 | return None |
| 1520 | |
| 1521 | target = skill_name.lower().strip() |
| 1522 | for root in visible_roots: |
| 1523 | for skill_md in discover_skill_md_files(Path(root)): |
| 1524 | skill = skill_from_markdown(skill_md, include_content=True) |
| 1525 | if not skill: |
| 1526 | continue |
| 1527 | candidates = { |
| 1528 | (skill.name or "").strip().lower(), |
| 1529 | skill.path.name.strip().lower(), |
| 1530 | } |
| 1531 | if target in candidates: |
| 1532 | return skill |
| 1533 | |
| 1534 | return None |
| 1535 | |
| 1536 | |
| 1537 | def _load_skill_from_runtime_path( |
| 1538 | skill_path: str, |
| 1539 | visible_roots: list[str], |
| 1540 | ) -> Skill | None: |
| 1541 | abs_path = files.fix_dev_path(skill_path) |
| 1542 | if not any(files.is_in_dir(abs_path, root) for root in visible_roots): |
| 1543 | return None |
| 1544 | |
| 1545 | skill_md = Path(abs_path) / "SKILL.md" |
| 1546 | if not skill_md.is_file(): |
| 1547 | return None |
| 1548 | |
| 1549 | return skill_from_markdown(skill_md, include_content=True) |
| 1550 | |
| 1551 | |
| 1552 | def _skill_entry(skill: Skill) -> ActiveSkillEntry: |
| 1553 | return { |
| 1554 | "name": skill.name or skill.path.name, |
| 1555 | "path": files.normalize_a0_path(str(skill.path)), |
| 1556 | } |
| 1557 | |
| 1558 | |
| 1559 | def _skill_matches_entries( |
| 1560 | skill: Skill, |
| 1561 | entries: list[ActiveSkillEntry], |
| 1562 | ) -> bool: |
| 1563 | skill_entry = _skill_entry(skill) |
| 1564 | return any(_entries_match(skill_entry, entry) for entry in entries) |
| 1565 | |
| 1566 | |
| 1567 | def _skill_is_hidden_for_agent(agent: Agent | None, skill: Skill) -> bool: |
| 1568 | if not agent: |
| 1569 | return False |
| 1570 | return _skill_matches_entries( |
| 1571 | skill, get_hidden_skills(agent) |
| 1572 | ) or not is_skill_allowed(get_visibility_policy(agent), skill) |
| 1573 | |
| 1574 | |
| 1575 | def _filter_hidden_skills( |
| 1576 | agent: Agent | None, |
| 1577 | skills: list[Skill], |
| 1578 | ) -> list[Skill]: |
| 1579 | if not agent: |
| 1580 | return skills |
| 1581 | |
| 1582 | hidden_entries = get_hidden_skills(agent) |
| 1583 | visibility_policy = get_visibility_policy(agent) |
| 1584 | return [ |
| 1585 | skill |
| 1586 | for skill in skills |
| 1587 | if not _skill_matches_entries(skill, hidden_entries) |
| 1588 | and is_skill_allowed(visibility_policy, skill) |
| 1589 | ] |
| 1590 | |
| 1591 | |
| 1592 | def _skill_visibility_aliases( |
| 1593 | skill_or_entry: Skill | ActiveSkillEntry | str, |
| 1594 | ) -> set[str]: |
| 1595 | if isinstance(skill_or_entry, Skill): |
| 1596 | values = ( |
| 1597 | skill_or_entry.name, |
| 1598 | skill_or_entry.path.name, |
| 1599 | files.normalize_a0_path(str(skill_or_entry.path)), |
| 1600 | ) |
| 1601 | elif isinstance(skill_or_entry, dict): |
| 1602 | values = ( |
| 1603 | str(skill_or_entry.get("name") or ""), |
| 1604 | str(skill_or_entry.get("path") or ""), |
| 1605 | ) |
| 1606 | else: |
| 1607 | values = (str(skill_or_entry or ""),) |
| 1608 | |
| 1609 | aliases: set[str] = set() |
| 1610 | for value in values: |
| 1611 | fixed = value.strip().replace("\\", "/").rstrip("/") |
| 1612 | if not fixed: |
| 1613 | continue |
| 1614 | aliases.add(fixed.casefold()) |
| 1615 | if "/" in fixed: |
| 1616 | aliases.add(fixed.rsplit("/", 1)[-1].casefold()) |
| 1617 | return aliases |