|
| 1 | +"""SkillOpt-Sleep — resolve a discovered skill name to a local ``SKILL.md``. |
| 2 | +
|
| 3 | +Skill names observed in transcripts are untrusted strings, so resolution is |
| 4 | +deliberately narrow: a name is normalized, matched only inside documented local |
| 5 | +skill roots, and reported. Nothing here writes, edits, or creates files. |
| 6 | +
|
| 7 | +Resolution outcomes are distinguishable on purpose — ``missing`` (no root has |
| 8 | +the skill) is a different signal from ``ambiguous`` (several roots do) and from |
| 9 | +``rejected`` (the name itself is unusable), so callers can fall back to the |
| 10 | +existing managed-skill behavior instead of guessing. |
| 11 | +""" |
| 12 | +from __future__ import annotations |
| 13 | + |
| 14 | +import os |
| 15 | +import re |
| 16 | +from dataclasses import dataclass |
| 17 | +from typing import List, Sequence, Tuple |
| 18 | + |
| 19 | +SKILL_FILENAME = "SKILL.md" |
| 20 | + |
| 21 | +FOUND = "found" |
| 22 | +MISSING = "missing" |
| 23 | +AMBIGUOUS = "ambiguous" |
| 24 | +REJECTED = "rejected" |
| 25 | + |
| 26 | + |
| 27 | +@dataclass(frozen=True) |
| 28 | +class SkillResolution: |
| 29 | + """The outcome of resolving one skill name. Never a partial success.""" |
| 30 | + |
| 31 | + name: str |
| 32 | + status: str |
| 33 | + path: str = "" |
| 34 | + candidates: Tuple[str, ...] = () |
| 35 | + reason: str = "" |
| 36 | + |
| 37 | + @property |
| 38 | + def ok(self) -> bool: |
| 39 | + return self.status == FOUND |
| 40 | + |
| 41 | + |
| 42 | +def normalize_skill_name(name: object) -> str: |
| 43 | + """Return a usable skill directory name, or "" when the name is unusable. |
| 44 | +
|
| 45 | + Only a single path segment is accepted: no separators, no parent traversal, |
| 46 | + no absolute or home-relative paths, no control characters. The name is |
| 47 | + whitespace-trimmed but otherwise preserved, since skill directories are |
| 48 | + case- and punctuation-sensitive. |
| 49 | + """ |
| 50 | + if not isinstance(name, str): |
| 51 | + return "" |
| 52 | + candidate = name.strip() |
| 53 | + if not candidate or candidate in {os.curdir, os.pardir}: |
| 54 | + return "" |
| 55 | + if candidate.startswith("~"): |
| 56 | + return "" |
| 57 | + if os.path.isabs(candidate) or os.path.splitdrive(candidate)[0]: |
| 58 | + return "" |
| 59 | + if "/" in candidate or "\\" in candidate or os.sep in candidate: |
| 60 | + return "" |
| 61 | + if os.altsep and os.altsep in candidate: |
| 62 | + return "" |
| 63 | + if any(ord(ch) < 32 or ord(ch) == 127 for ch in candidate): |
| 64 | + return "" |
| 65 | + return candidate |
| 66 | + |
| 67 | + |
| 68 | +def _listdir(path: str) -> List[str]: |
| 69 | + """Sorted directory entries, or [] when the directory is absent/unreadable.""" |
| 70 | + try: |
| 71 | + return sorted(os.listdir(path)) |
| 72 | + except OSError: |
| 73 | + return [] |
| 74 | + |
| 75 | + |
| 76 | +def _version_sort_key(name: str) -> tuple: |
| 77 | + """Order version directory names newest-last, numerically where possible. |
| 78 | +
|
| 79 | + ``1.10.0`` must sort above ``1.9.0``, so the leading numeric segments |
| 80 | + compare as ints rather than as strings. Anything after the first |
| 81 | + non-numeric segment is a prerelease suffix (``2.0.0-beta``), and a bare |
| 82 | + release outranks any prerelease sharing its numeric prefix — comparing the |
| 83 | + segment lists alone would do the opposite, because a shorter list that is a |
| 84 | + prefix of a longer one sorts lower. The name is the final tie-break so the |
| 85 | + order is always total and deterministic. |
| 86 | + """ |
| 87 | + numeric: List[int] = [] |
| 88 | + suffix: List[str] = [] |
| 89 | + for part in re.split(r"[._\-+]", name): |
| 90 | + if part.isdigit() and not suffix: |
| 91 | + numeric.append(int(part)) |
| 92 | + else: |
| 93 | + suffix.append(part) |
| 94 | + return (numeric, 0 if suffix else 1, suffix, name) |
| 95 | + |
| 96 | + |
| 97 | +def _plugin_skills_root(plugin_dir: str) -> str: |
| 98 | + """The single skills root for one installed plugin, or "" if it has none. |
| 99 | +
|
| 100 | + Claude marketplace installs are versioned — |
| 101 | + ``<plugin>/<version>/skills`` — and several versions of the same plugin can |
| 102 | + be present at once. Returning each of them as a peer root would make an |
| 103 | + ordinary upgraded plugin resolve AMBIGUOUS, so exactly one root is chosen: |
| 104 | + the newest version, falling back to the legacy unversioned |
| 105 | + ``<plugin>/skills`` layout when no version directory carries skills. |
| 106 | + """ |
| 107 | + versioned = [] |
| 108 | + for entry in _listdir(plugin_dir): |
| 109 | + candidate = os.path.join(plugin_dir, entry, "skills") |
| 110 | + if os.path.isdir(candidate): |
| 111 | + versioned.append((_version_sort_key(entry), candidate)) |
| 112 | + if versioned: |
| 113 | + versioned.sort() |
| 114 | + return versioned[-1][1] |
| 115 | + |
| 116 | + legacy = os.path.join(plugin_dir, "skills") |
| 117 | + return legacy if os.path.isdir(legacy) else "" |
| 118 | + |
| 119 | + |
| 120 | +def skill_search_roots(cfg: object) -> List[str]: |
| 121 | + """Documented local skill roots for a config: user skills, then plugin cache. |
| 122 | +
|
| 123 | + ``<claude_home>/skills`` holds hand-written skills. Installed Claude Code |
| 124 | + plugins expose theirs under the plugin cache, in either the versioned |
| 125 | + marketplace layout ``plugins/cache/<marketplace>/<plugin>/<version>/skills`` |
| 126 | + or the legacy ``plugins/cache/<marketplace>/<plugin>/skills``. At most one |
| 127 | + root per installed plugin is returned, in that fixed precedence order. |
| 128 | + """ |
| 129 | + configured = str(getattr(cfg, "claude_home", "") or "").strip() |
| 130 | + if not configured: |
| 131 | + # Guard before abspath: os.path.abspath("") is the CWD, which would |
| 132 | + # silently search a tree well outside the documented ~/.claude root. |
| 133 | + return [] |
| 134 | + claude_home = os.path.abspath(os.path.expanduser(configured)) |
| 135 | + roots = [os.path.join(claude_home, "skills")] |
| 136 | + |
| 137 | + cache = os.path.join(claude_home, "plugins", "cache") |
| 138 | + for marketplace in _listdir(cache): |
| 139 | + plugins_dir = os.path.join(cache, marketplace) |
| 140 | + if not os.path.isdir(plugins_dir): |
| 141 | + continue |
| 142 | + for plugin in _listdir(plugins_dir): |
| 143 | + root = _plugin_skills_root(os.path.join(plugins_dir, plugin)) |
| 144 | + if root: |
| 145 | + roots.append(root) |
| 146 | + return [r for r in roots if os.path.isdir(r)] |
| 147 | + |
| 148 | + |
| 149 | +def _contained_skill_file(root: str, name: str) -> str: |
| 150 | + """Return the real ``SKILL.md`` path under ``root`` for ``name``, else "". |
| 151 | +
|
| 152 | + Symlinks are followed and then re-checked against the real root, so a skill |
| 153 | + directory or file that points outside the root is refused rather than read. |
| 154 | + """ |
| 155 | + try: |
| 156 | + real_root = os.path.realpath(root) |
| 157 | + skill_file = os.path.realpath(os.path.join(real_root, name, SKILL_FILENAME)) |
| 158 | + except OSError: |
| 159 | + return "" |
| 160 | + if not os.path.isfile(skill_file): |
| 161 | + return "" |
| 162 | + try: |
| 163 | + contained = os.path.commonpath([real_root, skill_file]) == real_root |
| 164 | + except ValueError: |
| 165 | + # Different drives / mixed path flavours (Windows): by definition the |
| 166 | + # file is not inside this root, so refuse it rather than crash. |
| 167 | + return "" |
| 168 | + if not contained: |
| 169 | + return "" |
| 170 | + return skill_file |
| 171 | + |
| 172 | + |
| 173 | +def resolve_skill(name: object, roots: Sequence[str]) -> SkillResolution: |
| 174 | + """Resolve ``name`` against ``roots`` without touching any skill content.""" |
| 175 | + normalized = normalize_skill_name(name) |
| 176 | + if not normalized: |
| 177 | + return SkillResolution( |
| 178 | + name=name if isinstance(name, str) else "", |
| 179 | + status=REJECTED, |
| 180 | + reason="skill name is empty or not a single safe path segment", |
| 181 | + ) |
| 182 | + |
| 183 | + matches: List[str] = [] |
| 184 | + for root in roots: |
| 185 | + found = _contained_skill_file(root, normalized) |
| 186 | + if found and found not in matches: |
| 187 | + matches.append(found) |
| 188 | + |
| 189 | + if not matches: |
| 190 | + return SkillResolution( |
| 191 | + name=normalized, |
| 192 | + status=MISSING, |
| 193 | + reason=f"no {SKILL_FILENAME} for this skill in the configured skill roots", |
| 194 | + ) |
| 195 | + if len(matches) > 1: |
| 196 | + return SkillResolution( |
| 197 | + name=normalized, |
| 198 | + status=AMBIGUOUS, |
| 199 | + candidates=tuple(matches), |
| 200 | + reason="several skill roots define this skill", |
| 201 | + ) |
| 202 | + return SkillResolution( |
| 203 | + name=normalized, status=FOUND, path=matches[0], candidates=tuple(matches) |
| 204 | + ) |
0 commit comments