diff --git a/tools/skills_sync.py b/tools/skills_sync.py index 2ab5adcec2..02e8553ad2 100644 --- a/tools/skills_sync.py +++ b/tools/skills_sync.py @@ -1,15 +1,11 @@ #!/usr/bin/env python3 -""" -Skills Sync -- Manifest-based seeding and updating of bundled skills. +"""Skills Sync -- manifest-based seeding and updating of bundled skills. -Copies bundled skills from the repo's skills/ directory into ~/.hermes/skills/, -tracking each synced skill's origin hash in ~/.hermes/skills/.bundled_manifest -(v2: one "skill_name:origin_hash" line each; v1 plain-name lines auto-migrate). - -Update logic: NEW skills are copied and recorded; EXISTING skills update only -when bundled changed AND the user copy still matches the origin hash (a differing -user copy is user-customized -> SKIP); skills the user DELETED are not re-added; -skills REMOVED upstream are cleaned from the manifest. +Copies repo skills/ into ~/.hermes/skills/, tracking each synced skill's origin +hash in .bundled_manifest (v2 "name:hash" lines; v1 plain names auto-migrate). +NEW skills are copied and recorded; EXISTING skills update only when bundled +changed AND the user copy still matches the origin hash (else user-customized -> +SKIP); user-DELETED skills are not re-added; upstream-REMOVED ones leave the manifest. """ import hashlib @@ -22,17 +18,16 @@ from dataclasses import dataclass, field from pathlib import Path from typing import Dict, Iterator, List, Optional, Set, Tuple -# Force stdout/stderr to UTF-8: on GBK-style Windows locales the default encoding -# can't represent the glyphs printed here (✓ ↑ →) and raises mid-run; install.ps1 -# also parses this script's stdout as UTF-8 and aborts on a GBK byte stream. +# Force UTF-8 stdout/stderr: GBK-style Windows locales can't encode the glyphs +# printed here (✓ ↑ →), and install.ps1 parses this script's stdout as UTF-8. for _stream in (sys.stdout, sys.stderr): - if hasattr(_stream, "reconfigure"): - try: - _stream.reconfigure(encoding="utf-8", errors="replace") - except (ValueError, TypeError): - pass + try: + _stream.reconfigure(encoding="utf-8", errors="replace") + except (AttributeError, ValueError, TypeError): + pass from hermes_constants import get_bundled_skills_dir, get_hermes_home, get_optional_skills_dir from agent.skill_utils import ESSENTIAL_SKILLS, is_excluded_skill_path +from tools.skill_usage import _read_skill_name, read_suppressed_names # noqa: F401 (re-exported) from utils import atomic_write_text logger = logging.getLogger(__name__) @@ -41,10 +36,9 @@ HERMES_HOME = get_hermes_home() SKILLS_DIR = HERMES_HOME / "skills" MANIFEST_FILE = SKILLS_DIR / ".bundled_manifest" -# Import-time snapshots backing the call-time accessors below. Long-lived -# multi-profile runtimes import this module once and later retarget HERMES_HOME -# via set_hermes_home_override(); frozen constants would then resolve (and for -# reset_bundled_skill() DELETE) against the wrong profile. The accessors honor +# Import-time snapshots backing the call-time accessors: long-lived multi-profile +# runtimes retarget HERMES_HOME after import, and frozen constants would resolve +# (and for reset_bundled_skill() DELETE) against the wrong profile. Accessors honor # an explicitly patched module global and otherwise re-resolve on every call. _HERMES_HOME_AT_IMPORT = HERMES_HOME _SKILLS_DIR_AT_IMPORT = SKILLS_DIR @@ -68,9 +62,8 @@ def _manifest_file() -> Path: return _live(MANIFEST_FILE, _MANIFEST_FILE_AT_IMPORT, lambda: _skills_dir() / ".bundled_manifest") -# Written by `hermes profile create --no-skills` / the installer's `--no-skills`; -# when present in HERMES_HOME, sync_skills() seeds only the essential skills. Mirrors -# hermes_cli.profiles.NO_BUNDLED_SKILLS_MARKER (literal: no CLI import in this module). +# Written by `hermes profile create --no-skills` / installer `--no-skills`: sync seeds +# only essential skills. Mirrors hermes_cli.profiles.NO_BUNDLED_SKILLS_MARKER (no CLI import here). NO_BUNDLED_SKILLS_MARKER = ".no-bundled-skills" @@ -125,20 +118,8 @@ def _read_manifest() -> Dict[str, str]: def _read_suppressed_names() -> set: - """Built-in skills the curator pruned — must NOT be re-seeded. Delegates to - ``tools.skill_usage`` (source of truth), falling back to reading ``.curator_suppressed`` - directly if that import is unavailable in a packaged/update context.""" - try: - from tools.skill_usage import read_suppressed_names - - return read_suppressed_names() - except Exception: - path = _skills_dir() / ".curator_suppressed" - try: - lines = path.read_text(encoding="utf-8").splitlines() if path.exists() else [] - except OSError: - return set() - return {line.strip() for line in lines if line.strip() and not line.strip().startswith("#")} + """Built-in skills the curator pruned — must NOT be re-seeded (tests patch this name).""" + return read_suppressed_names() def _write_manifest(entries: Dict[str, str]): @@ -152,35 +133,16 @@ def _write_manifest(entries: Dict[str, str]): logger.debug("Failed to write skills manifest %s: %s", _manifest_file(), e, exc_info=True) -def _read_skill_name(skill_md: Path, fallback: str) -> str: - """Read the name field from SKILL.md YAML frontmatter, falling back to *fallback*.""" - try: - content = skill_md.read_text(encoding="utf-8", errors="replace")[:4000] - except OSError: - return fallback - in_frontmatter = False - for stripped in map(str.strip, content.split("\n")): - if stripped == "---": - if in_frontmatter: - break - in_frontmatter = True - elif in_frontmatter and stripped.startswith("name:"): - if value := stripped.split(":", 1)[1].strip().strip("\"'"): - return value - return fallback - - def _discover_bundled_skills(bundled_dir: Path) -> List[Tuple[str, Path]]: - """``(skill_name, skill_dir)`` for every SKILL.md under the bundled dir. Exclusions - are evaluated relative to the bundled tree: the install prefix itself may contain - ``venv``/``site-packages``, which once made every wheel install discover zero skills.""" + """``(skill_name, skill_dir)`` per SKILL.md under the bundled dir. Exclusions are + evaluated relative to the bundled tree: the install prefix itself may contain + ``venv``/``site-packages`` (which once made wheel installs discover zero skills).""" if not bundled_dir.exists(): return [] return [ (_read_skill_name(md, md.parent.name), md.parent) for md in bundled_dir.rglob("SKILL.md") - if not is_excluded_skill_path(md.relative_to(bundled_dir), root=bundled_dir) - ] + if not is_excluded_skill_path(md.relative_to(bundled_dir), root=bundled_dir)] def _compute_relative_dest(skill_dir: Path, bundled_dir: Path) -> Path: @@ -226,12 +188,9 @@ def _copy_dir(src: Path, dest: Path) -> None: def _recover_renamed_skill(st: "_SyncState", skill_name: str, dest: Path) -> Optional[str]: """Move a bundled skill's stale copy to its new canonical path after an upstream - RENAME/RECATEGORIZATION (manifest key still matches, ``dest`` doesn't exist yet; - otherwise the skill is misread as user-deleted and the old dir stranded forever). - + RENAME/RECATEGORIZATION (else it is misread as user-deleted and stranded forever). Only a copy byte-identical to the origin hash — proof *we* placed it — is moved; - user-edited or hub-installed copies are left. Returns the rel source path on move. - """ + user-edited or hub-installed copies stay. Returns the rel source path on move.""" origin_hash = st.manifest.get(skill_name, "") if not origin_hash: return None @@ -249,13 +208,11 @@ def _recover_renamed_skill(st: "_SyncState", skill_name: str, dest: Path) -> Opt if rel in st.hub_paths: # the hub owns its install paths continue if _dir_hash(candidate) != origin_hash: - # Moving a customized copy would edit the user's work; leaving it - # avoids a duplicate-name collision. Warn so they migrate deliberately. + # Moving a customized copy would edit the user's work; warn so they migrate deliberately. st.say( f" ⚠ {skill_name}: upstream moved this skill to {_rel_skills_posix(dest)}, but your " f"modified copy at {rel} was kept — it will not receive updates. " - f"Run `hermes skills reset {skill_name} --restore` to move to the new location." - ) + f"Run `hermes skills reset {skill_name} --restore` to move to the new location.") continue try: _move_dir(candidate, dest) @@ -281,8 +238,7 @@ class _SyncState: suppressed: List[str] = field(default_factory=list) relocated: List[str] = field(default_factory=list) shadowed_by_external: List[str] = field(default_factory=list) - # Rename-recovery indexes are expensive on host bind mounts: built lazily, - # only when a tracked skill is actually missing from its canonical path. + # Rename-recovery indexes are expensive on bind mounts: built lazily, only when needed. active_index: Optional[Dict[str, List[Path]]] = None hub_paths: Set[str] = field(default_factory=set) @@ -306,8 +262,8 @@ def _recover_orphan_backup(dest: Path) -> None: def _defer_to_external(st: _SyncState, skill_name: str, dest: Path, bundled_hash: str) -> None: """An external_dirs source provides this skill; a local copy would be a name collision - the loader refuses to resolve. Defer for ALL manifest states and self-heal a stale local - shadow from an earlier sync — only when byte-identical (a user's own skill differs).""" + the loader refuses. Defer for ALL manifest states and remove a stale local shadow from + an earlier sync — only when byte-identical (a user's own skill differs).""" st.shadowed_by_external.append(skill_name) st.skipped += 1 st.say(f" ⇢ {skill_name} (deferred to external_dirs, not written to local tree)") @@ -322,8 +278,8 @@ def _install_new_skill(st: _SyncState, skill_name: str, skill_src: Path, dest: P try: if dest.exists(): # Never overwrite a same-named user skill. Baseline the manifest only when - # byte-identical to bundled: recording bundled_hash for a differing copy - # would read as "user-modified" forever and block every bundled update. + # byte-identical: recording bundled_hash for a differing copy would read as + # "user-modified" forever and block every bundled update. st.skipped += 1 if _dir_hash(dest) == bundled_hash: st.manifest[skill_name] = bundled_hash @@ -338,15 +294,14 @@ def _install_new_skill(st: _SyncState, skill_name: str, skill_src: Path, dest: P st.manifest[skill_name] = bundled_hash st.say(f" + {skill_name}") except (OSError, IOError) as e: - st.say(f" ! Failed to copy {skill_name}: {e}") - # Not added to manifest — next sync retries. + st.say(f" ! Failed to copy {skill_name}: {e}") # not in manifest — next sync retries def _replace_skill_dir(skill_src: Path, dest: Path) -> None: """Replace ``dest`` with a fresh copy of ``skill_src`` via a ``.bak`` sibling, restoring the original on failure.""" backup = dest.with_suffix(".bak") - if backup.exists(): # a stale .bak would make shutil.move() nest dest INSIDE it; dest is authoritative + if backup.exists(): # a stale .bak would make shutil.move() nest dest INSIDE it _rmtree_writable(backup) shutil.move(str(dest), str(backup)) try: @@ -375,8 +330,7 @@ def _update_existing_skill(st: _SyncState, skill_name: str, skill_src: Path, des return user_hash = _dir_hash(dest) if not origin_hash: - # v1 migration: baseline from the user's copy so future syncs can detect edits. - # Can't tell user-edit from upstream change — be safe and skip. + # v1 migration: baseline from the user's copy (can't tell user-edit from upstream change). st.manifest[skill_name] = user_hash st.skipped += 1 return @@ -410,11 +364,9 @@ def _seed_category_descriptions(bundled_dir: Path, only_dirs: Optional[Set[Path] def sync_skills(quiet: bool = False) -> dict: - """Sync bundled skills into ~/.hermes/skills/ using the manifest. Returns a dict - with keys copied, updated, skipped, user_modified, cleaned, suppressed, relocated, - total_bundled, optional_provenance_backfilled, shadowed_by_external, skipped_opt_out.""" - # Opted-out profiles seed ONLY ESSENTIAL_SKILLS: the system prompt always points at - # ``hermes-agent``, so even a Blank Slate profile keeps it. + """Sync bundled skills into ~/.hermes/skills/ using the manifest; returns the + per-category result dict (see the final return). Opted-out profiles seed ONLY + ESSENTIAL_SKILLS: the system prompt always points at ``hermes-agent``.""" essential_only = (_hermes_home() / NO_BUNDLED_SKILLS_MARKER).exists() if essential_only and not quiet: print(" (profile opted out of bundled skills via .no-bundled-skills — seeding essential skills only)") @@ -455,7 +407,7 @@ def sync_skills(quiet: bool = False) -> dict: st.skipped += 1 # in manifest but not on disk — user deleted it # Clean manifest entries for skills removed upstream. Skipped when opted out: bundled_skills - # is only the essential set there, so cleaning would drop tracking for every other skill. + # is only the essential set there, so cleaning would drop tracking for everything else. cleaned = [] if essential_only else sorted(set(st.manifest) - {name for name, _ in bundled_skills}) for name in cleaned: del st.manifest[name] @@ -477,14 +429,10 @@ def sync_skills(quiet: bool = False) -> dict: def _rmtree_writable(path: Path) -> None: - """Remove a directory tree, making read-only entries writable first (Nix/deb/rpm - sources keep r-x dirs; unlinking a child needs a writable parent, so chmod both). - - Scope guard: refuses anything not a STRICT child of the active profile's skills - root, so a bad path join / missing HERMES_HOME / malicious manifest entry raises - a loud ValueError instead of wiping ``~/.hermes``. Callers always pass a skill dir - or its ``.bak`` sibling; the skills root itself must never be removed. - """ + """rmtree that first makes read-only entries writable (Nix/deb/rpm keep r-x dirs; + unlinking a child needs a writable parent, so chmod both). Scope guard: refuses + anything not a STRICT child of the active skills root, so a bad join / missing + HERMES_HOME / malicious manifest entry raises instead of wiping ``~/.hermes``.""" target = Path(path).resolve() skills_root = _skills_dir().resolve() if skills_root not in target.parents: @@ -504,13 +452,11 @@ def _rmtree_writable(path: Path) -> None: # Re-exported so ``from tools.skills_sync import X`` / ``patch("tools.skills_sync.X")`` keep working. from tools.skills_sync_bundled_ops import ( # noqa: E402,F401 _is_tracked_user_modification, _read_for_diff, diff_bundled_skill, list_user_modified_bundled_skills, - remove_pristine_bundled_skills, reset_bundled_skill, set_bundled_skills_opt_out, -) + remove_pristine_bundled_skills, reset_bundled_skill, set_bundled_skills_opt_out) from tools.skills_sync_optional import ( # noqa: E402,F401 _backfill_optional_provenance, _content_hash, _index_installed_skill_dirs_by_name, _move_to_restore_backup, _optional_skill_index, _read_hub_install_paths, _safe_rel_install_path, _skill_file_list, - restore_official_optional_skill, -) + restore_official_optional_skill) if __name__ == "__main__": diff --git a/tools/skills_sync_bundled_ops.py b/tools/skills_sync_bundled_ops.py index 9e843a1d0b..7d8cc77822 100644 --- a/tools/skills_sync_bundled_ops.py +++ b/tools/skills_sync_bundled_ops.py @@ -1,24 +1,19 @@ """Bundled-skill maintenance ops: reset, diff, list-modified, opt-out, remove-pristine. -Extracted from ``tools.skills_sync``. Profile-scoped paths and patchable helpers -(``_get_bundled_dir``, ``sync_skills``, ...) are resolved through ``_ss()`` at -call time so monkeypatching ``tools.skills_sync`` keeps working. +Profile-scoped paths and patchable helpers (``_get_bundled_dir``, ``sync_skills``, ...) +are resolved through ``_ss()`` at call time so monkeypatching ``tools.skills_sync`` works. """ from pathlib import Path from typing import List, Optional, Tuple - -def _ss(): - from tools import skills_sync - - return skills_sync +from tools.skills_sync_optional import _skill_file_list, _ss def _is_tracked_user_modification(origin_hash: str, user_hash: str) -> bool: - """Whether an on-disk skill is a user modification ``hermes update`` keeps. Shared by - the sync loop and ``list_user_modified_bundled_skills`` so they never drift: needs a - recorded origin hash (un-baselined v1 entries don't count) AND differing content.""" + """User modification ``hermes update`` keeps: a recorded origin hash (un-baselined v1 + entries don't count) AND differing content. Shared by the sync loop and + ``list_user_modified_bundled_skills`` so they never drift.""" return bool(origin_hash) and user_hash != origin_hash @@ -29,12 +24,10 @@ def _bundled_by_name(bundled_dir: Path) -> dict: def reset_bundled_skill(name: str, restore: bool = False) -> dict: """Reset a bundled skill's manifest tracking so future syncs work normally. - An edited bundled skill stays ``user_modified`` forever — even after copying the - bundled version back — because the manifest holds the OLD origin hash; clearing - the entry breaks that loop. ``restore`` also deletes the user's copy so the next - sync re-copies bundled. Returns ``{ok, action, message, synced}``; action is - manifest_cleared / restored / not_in_manifest / bundled_missing / not_reset. - """ + An edited bundled skill stays ``user_modified`` forever because the manifest holds + the OLD origin hash; clearing the entry breaks that loop. ``restore`` also deletes + the user's copy so the next sync re-copies bundled. Returns ``{ok, action, message, + synced}``; action is manifest_cleared / restored / not_in_manifest / bundled_missing / not_reset.""" ss = _ss() manifest = ss._read_manifest() bundled_dir = ss._get_bundled_dir() @@ -49,8 +42,8 @@ def reset_bundled_skill(name: str, restore: bool = False) -> dict: return _fail("not_in_manifest", f"'{name}' is not a tracked bundled skill. Nothing to reset. " f"(Hub-installed skills use `hermes skills uninstall`.)") - # Delete the user's copy BEFORE touching the manifest so a failed rmtree - # cannot leave the skill in a manifest-less limbo state. + # Delete the user's copy BEFORE touching the manifest so a failed rmtree can't + # leave the skill in a manifest-less limbo. deleted_user_copy = False if restore: if not is_bundled: @@ -115,8 +108,6 @@ def diff_bundled_skill(name: str) -> dict: modified / added (only in user copy) / removed (only in bundled) / binary.""" import difflib - from tools.skills_sync_optional import _skill_file_list - ss = _ss() def _fail(found: bool, message: str) -> dict: @@ -149,8 +140,7 @@ def diff_bundled_skill(name: str) -> dict: elif user_text != stock_text: text = "".join(difflib.unified_diff( stock_text.splitlines(keepends=True), user_text.splitlines(keepends=True), - fromfile=f"stock/{rel}", tofile=f"yours/{rel}", - )) + fromfile=f"stock/{rel}", tofile=f"yours/{rel}")) diffs.append({"path": rel, "status": "modified", "diff": text}) message = (f"'{name}' differs from the stock version in {len(diffs)} file(s)." if diffs @@ -162,14 +152,12 @@ _OPT_OUT_MESSAGES = { # (enabled, changed) -> message (True, True): "Opted out of bundled skills. Future install / update / sync runs will not seed bundled skills into this profile.", (True, False): "Already opted out — marker was already present.", (False, True): "Opted back in. The next `hermes update` (or `hermes skills opt-in --sync`) will re-seed bundled skills.", - (False, False): "Not opted out — no marker to remove.", -} + (False, False): "Not opted out — no marker to remove."} def set_bundled_skills_opt_out(enabled: bool) -> dict: - """Toggle the .no-bundled-skills marker: the on-disk half of ``hermes skills - opt-out`` / ``opt-in`` that stops installer/update/sync seeding. Removing - already-present skills is a separate step (``remove_pristine_bundled_skills``). + """Toggle the .no-bundled-skills marker (the on-disk half of ``hermes skills + opt-out`` / ``opt-in``); removing present skills is ``remove_pristine_bundled_skills``. Returns ``{ok, changed, marker, message}``.""" ss = _ss() marker = ss._hermes_home() / ss.NO_BUNDLED_SKILLS_MARKER @@ -180,8 +168,7 @@ def set_bundled_skills_opt_out(enabled: bool) -> dict: marker.write_text( "This profile opted out of bundled-skill seeding (`hermes skills opt-out`).\n" "Delete this file to re-enable sync on the next `hermes update`.\n", - encoding="utf-8", - ) + encoding="utf-8") elif existed: marker.unlink() except OSError as e: @@ -192,14 +179,10 @@ def set_bundled_skills_opt_out(enabled: bool) -> dict: def remove_pristine_bundled_skills(dry_run: bool = False) -> dict: - """Delete bundled skills that are present, manifest-tracked, AND unmodified. - - Removed ONLY when in the sync manifest (genuinely bundled, not hub/hand-written), - still in the bundled source (hash-comparable), and byte-identical to the origin - hash; everything else lands in ``skipped``. Removed skills lose their manifest - entry so a later opt-in re-seed treats them as new. - Returns ``{ok, removed, skipped: [{name, reason}], dry_run, message}``. - """ + """Delete bundled skills that are manifest-tracked (genuinely bundled), still in the + bundled source (hash-comparable) AND byte-identical to the origin hash; everything + else lands in ``skipped``. Removed skills lose their manifest entry so a later + opt-in re-seed treats them as new. Returns ``{ok, removed, skipped: [{name, reason}], dry_run, message}``.""" ss = _ss() manifest = ss._read_manifest() bundled_dir = ss._get_bundled_dir() @@ -214,8 +197,7 @@ def remove_pristine_bundled_skills(dry_run: bool = False) -> dict: continue dest = ss._compute_relative_dest(src, bundled_dir) if not dest.exists(): - # Already gone from disk; just forget the stale manifest entry. - if not dry_run: + if not dry_run: # already gone from disk; forget the stale manifest entry manifest.pop(name, None) continue if ss._dir_hash(dest) != origin_hash: @@ -236,5 +218,4 @@ def remove_pristine_bundled_skills(dry_run: bool = False) -> dict: verb = "Would remove" if dry_run else "Removed" return { "ok": True, "removed": removed, "skipped": skipped, "dry_run": dry_run, - "message": f"{verb} {len(removed)} pristine bundled skill(s); kept {len(skipped)}.", - } + "message": f"{verb} {len(removed)} pristine bundled skill(s); kept {len(skipped)}."} diff --git a/tools/skills_sync_optional.py b/tools/skills_sync_optional.py index 91d0ef0ad7..93d9de09d7 100644 --- a/tools/skills_sync_optional.py +++ b/tools/skills_sync_optional.py @@ -1,8 +1,7 @@ """Official optional-skill provenance: hub-lock backfill and restore. -Extracted from ``tools.skills_sync``. Profile-scoped paths and patchable -helpers are resolved through ``_ss()`` at call time so tests and multi-profile -runtimes that patch ``tools.skills_sync`` globals keep working. +Profile-scoped paths and patchable helpers are resolved through ``_ss()`` at call +time so tests and multi-profile runtimes patching ``tools.skills_sync`` keep working. """ import json @@ -66,24 +65,26 @@ def _hub_lock_entries(data: Optional[dict]) -> List[dict]: def _read_hub_install_paths() -> Set[str]: - """Install paths recorded in the hub lock, as POSIX strings. Hub-installed skills - are owned by the hub, never by bundled sync: rename recovery must not move them even - when content matches a bundled origin hash, or the lock's ``install_path`` dangles.""" + """Hub-lock install paths as POSIX strings. Hub-installed skills are owned by the + hub: rename recovery must not move them even when content matches a bundled + origin hash, or the lock's ``install_path`` dangles.""" return {str(e["install_path"]).strip("/") for e in _hub_lock_entries(_load_hub_lock()) if e.get("install_path")} def _write_hub_lock(lock_path: Path, data: dict) -> None: - """Atomic write so a crash mid-write can't wipe all provenance (the - JSONDecodeError fallback in the reader resets ``installed`` to empty).""" + """Atomic: a crash mid-write must not wipe all provenance (the reader's + JSONDecodeError fallback resets ``installed`` to empty).""" atomic_write_text(lock_path, json.dumps(data, indent=2, ensure_ascii=False) + "\n", tmp_prefix=".lock_") def _iter_optional_skills(optional_dir: Path, *, root_relative: bool) -> Iterator[Tuple[Path, Path, str]]: """Yield ``(skill_md, src, install_path)`` for every safe official optional skill.""" for skill_md in sorted(optional_dir.rglob("SKILL.md")): - if root_relative and is_excluded_skill_path(skill_md.relative_to(optional_dir), root=optional_dir): - continue - if not root_relative and is_excluded_skill_path(skill_md): + if root_relative: + excluded = is_excluded_skill_path(skill_md.relative_to(optional_dir), root=optional_dir) + else: + excluded = is_excluded_skill_path(skill_md) + if excluded: continue try: yield skill_md, skill_md.parent, _safe_rel_install_path(skill_md.parent, optional_dir) @@ -92,9 +93,8 @@ def _iter_optional_skills(optional_dir: Path, *, root_relative: bool) -> Iterato def _optional_skill_index() -> Dict[str, Tuple[str, str, Path]]: - """Official optional skills keyed by BOTH folder name and frontmatter name, so callers - may pass either the hub-lock slug or the user-facing name. Values are - ``(folder_name, install_path, source_dir)``.""" + """Official optional skills keyed by BOTH folder name and frontmatter name (hub-lock + slug or user-facing name). Values are ``(folder_name, install_path, source_dir)``.""" ss = _ss() optional_dir = ss._get_optional_dir() index: Dict[str, Tuple[str, str, Path]] = {} @@ -133,9 +133,8 @@ def _find_active_copies(folder_name: str, src_frontmatter: str, dest: Path) -> L def restore_official_optional_skill(name: str, *, restore: bool = False) -> dict: """Restore one or all official optional skills from repo source. ``restore=False`` - only performs exact-match provenance backfill; ``restore=True`` repairs mutated / - reorganized skills by backing up matching active copies and copying the official - source into its canonical path.""" + only backfills exact-match provenance; ``restore=True`` also backs up matching + active copies and copies the official source into its canonical path.""" ss = _ss() def _fail(message: str) -> dict: @@ -172,8 +171,7 @@ def restore_official_optional_skill(name: str, *, restore: bool = False) -> dict return { "ok": True, "message": "Official optional skill repair complete.", "restored": restored, "backfilled": _backfill_optional_provenance(quiet=True), "backed_up": backed_up, - "backup_dir": str(backup_root) if backed_up else "", - } + "backup_dir": str(backup_root) if backed_up else ""} def _index_installed_skill_dirs_by_name() -> Dict[str, List[Path]]: @@ -192,10 +190,9 @@ def _index_installed_skill_dirs_by_name() -> Dict[str, List[Path]]: def _relocated_dest(src_name: str, index: Dict[str, List[Path]]) -> Optional[Tuple[Path, str]]: - """The active tree may hold a skill under a DIFFERENT category path than the - repo (upstream reorganizes; the installed copy keeps its old location). Fall - back to a UNIQUE same-directory-name match — an ambiguous name gives no basis - to pick one. Returns ``(dest, install_path)`` or None.""" + """The active tree may hold a skill under a DIFFERENT category path than the repo + (upstream reorganized; the installed copy kept its location). Fall back to a UNIQUE + same-directory-name match; ambiguity gives no basis to pick. ``(dest, install_path)`` or None.""" candidates = index.get(src_name, []) if len(candidates) != 1: return None @@ -208,9 +205,9 @@ def _relocated_dest(src_name: str, index: Dict[str, List[Path]]) -> Optional[Tup def _backfill_optional_provenance(quiet: bool = False) -> List[str]: - """Mark already-present official optional skills as hub-installed: skills that used - to be bundled (or were hand-copied) and now live under optional-skills/ get official - provenance when byte-identical to the source. Modified/local skills are left alone.""" + """Mark already-present official optional skills as hub-installed: formerly bundled + (or hand-copied) skills now under optional-skills/ get official provenance when + byte-identical to the source. Modified/local skills are left alone.""" ss = _ss() optional_dir = ss._get_optional_dir() if not optional_dir.exists(): @@ -250,8 +247,7 @@ def _backfill_optional_provenance(quiet: bool = False) -> List[str]: "files": _skill_file_list(dest), "metadata": {"backfilled_from": "optional-skills"}, "installed_at": timestamp, - "updated_at": timestamp, - } + "updated_at": timestamp} existing_paths.add(install_path) backfilled.append(lock_name) if not quiet: diff --git a/tools/skills_tool.py b/tools/skills_tool.py index dded25917e..28767c86df 100644 --- a/tools/skills_tool.py +++ b/tools/skills_tool.py @@ -1,78 +1,60 @@ #!/usr/bin/env python3 -"""Skills Tool Module — list and view skill documents (progressive disclosure). +"""Skills Tool — list and view skill documents (progressive disclosure). -Skills are directories holding a SKILL.md (YAML frontmatter + instructions) plus -optional references/, templates/, assets/, scripts/ (agentskills.io layout). -Tier 1 (`skills_list`) returns name/description only; tier 2-3 (`skill_view`) -returns full content and linked files on demand. - -Frontmatter: name (≤64), description (≤1024) required; optional version, -license, platforms [macos|linux|windows], prerequisites.env_vars (legacy — -normalized into required_environment_variables), prerequisites.commands -(advisory only), compatibility, metadata.hermes.{tags,related_skills}. - -Sibling modules (every name re-exported here so ``from tools.skills_tool import X`` -and ``patch("tools.skills_tool.X")`` keep working): -- tools.skills_tool_setup: required env vars / secret capture / setup notes -- tools.skills_tool_plugin: plugin-skill serving + shared file/JSON helpers -- tools.skills_tool_dedup: repeat-view dedup registry +A skill is a directory holding SKILL.md (YAML frontmatter + instructions) plus +optional references/, templates/, assets/, scripts/. Tier 1 (`skills_list`) +returns name/description only; tier 2-3 (`skill_view`) returns full content +and linked files on demand. Sibling modules (skills_tool_setup / _plugin / +_dedup) re-export every name here so ``from tools.skills_tool import X`` and +``patch("tools.skills_tool.X")`` keep working. """ import json import logging import os import time +from contextlib import suppress +from dataclasses import dataclass from pathlib import Path, PurePosixPath, PureWindowsPath -from typing import Any, Dict, List, Optional, Set, Tuple +from typing import Any, Dict, List, Optional, Tuple from hermes_constants import get_hermes_home from tools.registry import registry, tool_error from hermes_cli.config import cfg_get from agent.skill_utils import ( - EXCLUDED_SKILL_DIRS as _EXCLUDED_SKILL_DIRS, - is_skill_support_path as _is_skill_support_path, -) + EXCLUDED_SKILL_DIRS as _EXCLUDED_SKILL_DIRS, is_skill_support_path as _is_skill_support_path) from tools.skills_tool_setup import ( # noqa: F401 SkillReadinessStatus, _ENV_VAR_NAME_RE, _REMOTE_ENV_BACKENDS, _build_setup_note, _capture_required_environment_variables, _collect_prerequisite_values, _gateway_setup_hint, _get_required_environment_variables, _get_terminal_backend_name, _is_env_var_persisted, _is_gateway_surface, _is_remote_env_backend, _normalize_prerequisite_values, _normalize_setup_metadata, - _remaining_required_environment_names, -) + _remaining_required_environment_names) from tools.skills_tool_plugin import ( # noqa: F401 MAX_DESCRIPTION_LENGTH, MAX_NAME_LENGTH, _INJECTION_PATTERNS, _available_skill_files, _fail, _json, _mark_background_review_read, _plugin_skill_linked_files, _preprocess_skill, _read_skill_text, _serve_plugin_skill, _serve_skill_file, - _truncate_description, -) + _truncate_description) from tools.skills_tool_dedup import ( # noqa: F401 _SKILL_VIEW_DEDUP_CAP, _SKILL_VIEW_DEDUP_MESSAGE, _check_skill_view_dedup, _record_skill_view, _skill_view_fingerprint, _skill_view_tracker, - _skill_view_tracker_lock, reset_skill_view_dedup, -) + _skill_view_tracker_lock, reset_skill_view_dedup) logger = logging.getLogger(__name__) -# Per-session skill discovery cache (re-reading every SKILL.md per call is -# wasteful with hundreds of skills). Validation mirrors -# hermes_cli/profiles.py::_count_skills (d5eee133e): the signature is the -# per-dir max mtime of the dir AND its immediate children (catches add/remove -# inside categories, which does NOT bump the root mtime) plus the disabled-set -# (config-driven, no mtime bump at all); a short TTL bounds staleness from -# in-place SKILL.md edits, invisible to any directory signature. -# skip_disabled True/False are cached separately. -_SKILLS_CACHE: dict = {} # {cache_key: (signature, timestamp, skills_list)} +# Per-session discovery cache: {cache_key: (signature, timestamp, skills_list)}. +# Signature = per-dir max mtime of the dir and its immediate children (add/remove +# inside a category does NOT bump the root mtime) + the disabled set (config-only +# change, no mtime at all) + platform; the TTL bounds staleness from in-place +# SKILL.md edits, which no directory signature can see. +_SKILLS_CACHE: dict = {} _SKILLS_CACHE_TTL_SECONDS = 30.0 -_SKILLS_CACHE_KEY_DISABLED = "with_disabled" -_SKILLS_CACHE_KEY_FILTERED = "filtered" def _skills_scan_signature(dirs_to_scan, disabled) -> tuple: - """Cheap change-signature for the skill scan inputs: O(#dirs + #categories) - stat calls, not a recursive walk. Includes the platform the scan's - ``skill_matches_platform`` filter will use (read from ``agent.skill_utils``'s - ``sys`` so test patches are honored) — the scan result is platform-dependent.""" + """O(#dirs + #categories) stat-based change signature; includes the platform + ``skill_matches_platform`` will use (via ``agent.skill_utils.sys`` so test + patches are honored) because the scan result is platform-dependent.""" from agent import skill_utils as _skill_utils platform = getattr(getattr(_skill_utils, "sys", None), "platform", "") @@ -82,39 +64,27 @@ def _skills_scan_signature(dirs_to_scan, disabled) -> tuple: m = d.stat().st_mtime except OSError: continue - try: - with os.scandir(d) as it: - for entry in it: - try: - if entry.is_dir(follow_symlinks=False): - m = max(m, entry.stat(follow_symlinks=False).st_mtime) - except OSError: - continue - except OSError: - pass + with suppress(OSError), os.scandir(d) as it: + for entry in it: + with suppress(OSError): + if entry.is_dir(follow_symlinks=False): + m = max(m, entry.stat(follow_symlinks=False).st_mtime) sig.append((str(d), m)) return (tuple(sig), frozenset(disabled), platform) -# All skills live in ~/.hermes/skills/ (seeded from bundled skills/ on install): -# agent edits, hub installs, and bundled skills coexist here without polluting -# the git repo. +# All skills live in ~/.hermes/skills/ (seeded from bundled skills/ on install). HERMES_HOME = get_hermes_home() SKILLS_DIR = HERMES_HOME / "skills" _SKILLS_DIR_AT_IMPORT = SKILLS_DIR def _skills_dir() -> Path: - """Active profile's skills directory at call time. - - Long-lived runtimes may import this module before the active profile has set - HERMES_HOME. Keep the legacy SKILLS_DIR attribute for tests and external - patchers, but when it has not been patched, resolve from the live - profile-scoped HERMES_HOME on every call.""" + """Active profile's skills dir at call time: the patched ``SKILLS_DIR`` when a + test/external patcher changed it, else the live profile-scoped HERMES_HOME + (long-lived runtimes may import this module before the profile is set).""" configured = Path(SKILLS_DIR) - if configured != _SKILLS_DIR_AT_IMPORT: - return configured - return get_hermes_home() / "skills" + return configured if configured != _SKILLS_DIR_AT_IMPORT else get_hermes_home() / "skills" _secret_capture_callback = None @@ -122,13 +92,9 @@ _LOOKUP_HINT = "Use a skill name or relative path within the skills directory." def _skill_lookup_path_error(name: str) -> Optional[str]: - """Error if a local skill lookup *name* can escape the search roots. - - ``name`` is joined onto each trusted search dir, so it must stay relative and - free of ``..`` — otherwise ``"../outside"`` or an absolute path could select a - skill (and read files) outside the skills directory. Mirrors the ``file_path`` - validation in ``tools.path_security``. Windows drive paths (``C:\\skills``) are - rejected too: their ``:`` would be misread as a plugin namespace separator.""" + """Error if lookup *name* could escape the search roots it is joined onto. + Windows drive paths are rejected too: their ``:`` would be misread as a + plugin namespace separator.""" from tools.path_security import has_traversal_component if not isinstance(name, str): @@ -148,8 +114,7 @@ def load_env() -> Dict[str, str]: env_vars: Dict[str, str] = {} if not env_path.exists(): return env_vars - # utf-8-sig: users hand-edit .env in Notepad, whose BOM would otherwise glue - # U+FEFF onto the first key (same dialect as hermes_cli/config.py readers). + # utf-8-sig: a Notepad BOM would otherwise glue U+FEFF onto the first key. with env_path.open(encoding="utf-8-sig", errors="replace") as f: for line in f: line = line.strip() @@ -166,31 +131,22 @@ def set_secret_capture_callback(callback) -> None: _secret_capture_callback = callback -# Lazy delegates to ``agent.skill_utils`` — public re-exports so existing -# callers (and tests patching either module) don't need updating. -def skill_matches_platform(frontmatter: Dict[str, Any]) -> bool: - """Check if a skill is compatible with the current OS platform.""" - from agent.skill_utils import skill_matches_platform as _impl - return _impl(frontmatter) +def _skill_utils_delegate(attr: str): + """Lazy call-time delegate to ``agent.skill_utils.`` — public re-exports so + existing callers (and tests patching either module) don't need updating.""" + def _delegate(*args): + from agent import skill_utils + return getattr(skill_utils, attr)(*args) + _delegate.__name__ = _delegate.__qualname__ = attr + return _delegate -def skill_matches_environment(frontmatter: Dict[str, Any]) -> bool: - """Offer-time relevance gate (kanban/docker/s6), NOT a hard-compatibility - gate; explicit skill loads bypass it.""" - from agent.skill_utils import skill_matches_environment as _impl - return _impl(frontmatter) - - -def _parse_frontmatter(content: str) -> Tuple[Dict[str, Any], str]: - """Parse YAML frontmatter from markdown content.""" - from agent.skill_utils import parse_frontmatter - return parse_frontmatter(content) - - -def _get_disabled_skill_names() -> Set[str]: - """Load disabled skill names from config.""" - from agent.skill_utils import get_disabled_skill_names - return get_disabled_skill_names() +skill_matches_platform = _skill_utils_delegate("skill_matches_platform") +# Offer-time relevance gate (kanban/docker/s6), NOT a hard-compatibility gate; +# explicit skill loads bypass it. +skill_matches_environment = _skill_utils_delegate("skill_matches_environment") +_parse_frontmatter = _skill_utils_delegate("parse_frontmatter") +_get_disabled_skill_names = _skill_utils_delegate("get_disabled_skill_names") def check_skills_requirements() -> bool: @@ -199,28 +155,23 @@ def check_skills_requirements() -> bool: def _get_category_from_path(skill_path: Path) -> Optional[str]: - """``~/.hermes/skills/mlops/axolotl/SKILL.md`` -> ``"mlops"`` (also for - skills.external_dirs). Tries the active profile dir first (respects test - monkeypatching), then external dirs from config.""" + """``~/.hermes/skills/mlops/axolotl/SKILL.md`` -> ``"mlops"``; active profile + dir first (respects test monkeypatching), then skills.external_dirs.""" dirs_to_check = [_skills_dir()] - try: + with suppress(Exception): from agent.skill_utils import get_external_skills_dirs dirs_to_check.extend(get_external_skills_dirs()) - except Exception: - pass for skills_dir in dirs_to_check: - try: + with suppress(ValueError): parts = skill_path.relative_to(skills_dir).parts - except ValueError: - continue - if len(parts) >= 3: - return parts[0] + if len(parts) >= 3: + return parts[0] return None def _parse_tags(tags_value) -> List[str]: - """Tags from frontmatter: a parsed list (yaml.safe_load), a bracket-wrapped - string "[a, b]", or a comma-separated string "a, b".""" + """Tags from frontmatter: a parsed list, a bracket-wrapped string "[a, b]", + or a comma-separated string "a, b".""" if not tags_value: return [] if isinstance(tags_value, list): @@ -232,9 +183,8 @@ def _parse_tags(tags_value) -> List[str]: def _get_session_platform() -> str: - """Platform from gateway session context. Mirrors - ``agent.skill_utils.get_disabled_skill_names`` so ``_is_skill_disabled`` - respects ``HERMES_SESSION_PLATFORM``.""" + """Platform from gateway session context (mirrors + ``agent.skill_utils.get_disabled_skill_names``).""" try: from gateway.session_context import get_session_env return get_session_env("HERMES_SESSION_PLATFORM") or "" @@ -243,8 +193,10 @@ def _get_session_platform() -> str: def _is_skill_disabled(name: str, platform: str = None) -> bool: - """Is the skill disabled in config? Platform precedence: explicit arg, - ``HERMES_PLATFORM`` env, ``HERMES_SESSION_PLATFORM`` from session context.""" + """Disabled in config? Platform precedence: explicit arg, ``HERMES_PLATFORM`` + env, ``HERMES_SESSION_PLATFORM`` from session context. A globally-disabled + skill stays disabled on every platform; the platform list adds to it (keep in + sync with agent.skill_utils.get_disabled_skill_names).""" try: from hermes_cli.config import load_config skills_cfg = load_config().get("skills", {}) @@ -253,9 +205,6 @@ def _is_skill_disabled(name: str, platform: str = None) -> bool: if resolved_platform: platform_disabled = cfg_get(skills_cfg, "platform_disabled", resolved_platform) if platform_disabled is not None: - # A globally-disabled skill stays disabled on every platform; the - # platform list adds to it. Keep in sync with - # agent.skill_utils.get_disabled_skill_names. return name in platform_disabled or name in global_disabled return name in global_disabled except Exception: @@ -263,12 +212,9 @@ def _is_skill_disabled(name: str, platform: str = None) -> bool: def _skill_search_dirs() -> Tuple[list, list, Path]: - """(project_dirs, all_dirs, active_skills_dir) in precedence order. - - Trusted project-local dirs come FIRST: first-wins dedup / the collision - resolver give them precedence over same-named local/external skills. - ``_skills_dir()`` resolves the LIVE profile HERMES_HOME; module-level - SKILLS_DIR can be stale in long-lived runtimes.""" + """(project_dirs, all_dirs, active_skills_dir). Trusted project-local dirs come + FIRST so first-wins dedup / the collision resolver prefer them over + same-named local/external skills.""" from agent.skill_utils import get_external_skills_dirs, get_project_skills_dirs project_dirs = list(get_project_skills_dirs()) @@ -281,17 +227,12 @@ def _skill_search_dirs() -> Tuple[list, list, Path]: def _find_all_skills(*, skip_disabled: bool = False) -> List[Dict[str, Any]]: - """All skills (name, description, category) in ~/.hermes/skills/ and - project/external dirs. ``skip_disabled=True`` returns ALL skills regardless - of disabled state (used by the ``hermes skills`` config UI). - - Cached per-session; invalidated when the scan signature changes (dir/category - mtimes or the disabled-set) and after a short TTL (in-place SKILL.md edits).""" + """All skills (name, description, category) across local/project/external dirs, + first-wins by name. ``skip_disabled=True`` returns ALL skills regardless of + disabled state (used by the ``hermes skills`` config UI). Cached per session.""" from agent.skill_utils import iter_project_skill_files, iter_skill_index_files - cache_key = _SKILLS_CACHE_KEY_DISABLED if skip_disabled else _SKILLS_CACHE_KEY_FILTERED - # Disabled set loaded once (not per-skill); part of the cache signature since - # disabling a skill is a config change with no filesystem mtime bump. + cache_key = "with_disabled" if skip_disabled else "filtered" disabled = set() if skip_disabled else _get_disabled_skill_names() project_dirs, dirs_to_scan, _ = _skill_search_dirs() signature = _skills_scan_signature(dirs_to_scan, disabled) @@ -299,15 +240,13 @@ def _find_all_skills(*, skip_disabled: bool = False) -> List[Dict[str, Any]]: cached = _SKILLS_CACHE.get(cache_key) if cached is not None and cached[0] == signature and (now - cached[1]) < _SKILLS_CACHE_TTL_SECONDS: - # Per-call shallow copies: callers mutate the returned dicts (web_server - # annotates s["enabled"]/s["usage"]); handing out cached objects would - # poison the cache for everyone else. + # Shallow copies: callers mutate the returned dicts (web_server annotates + # s["enabled"]/s["usage"]); handing out cached objects would poison the cache. return [dict(s) for s in cached[2]] skills = [] seen_names: set = set() - # Project dirs first, then local, then external (first-wins). Project dirs - # iterate through the quarantine chokepoint (scan-time injection gate). + # Project dirs iterate through the quarantine chokepoint (scan-time injection gate). for scan_dir in dirs_to_scan: _iter = iter_project_skill_files if scan_dir in project_dirs else lambda d: iter_skill_index_files(d, "SKILL.md") for skill_md in _iter(scan_dir): @@ -331,16 +270,14 @@ def _find_all_skills(*, skip_disabled: bool = False) -> List[Dict[str, Any]]: skills.append({ "name": name, "description": _truncate_description(description), - "category": _get_category_from_path(skill_md), - }) + "category": _get_category_from_path(skill_md)}) except (UnicodeDecodeError, PermissionError) as e: logger.debug("Failed to read skill file %s: %s", skill_md, e) except Exception as e: logger.debug("Skipping skill at %s: failed to parse: %s", skill_md, e, exc_info=True) # Keyed by the signature computed BEFORE the scan: a write racing the scan - # changes the signature, so the next call re-scans instead of serving the - # torn result past the TTL. Same shallow-copy contract as the hit path. + # changes the signature, so the next call re-scans instead of serving a torn result. _SKILLS_CACHE[cache_key] = (signature, now, skills) return [dict(s) for s in skills] @@ -352,7 +289,7 @@ def _sort_skills(skills: List[Dict[str, Any]]) -> List[Dict[str, Any]]: def skills_list(category: str = None, task_id: str = None) -> str: """Tier 1 listing: name + description (+ category) only, to minimize tokens. - ``category`` filters; ``task_id`` is accepted for handler parity.""" + ``task_id`` is accepted for handler parity.""" try: active_skills_dir = _skills_dir() if not active_skills_dir.exists(): @@ -374,8 +311,7 @@ def skills_list(category: str = None, task_id: str = None) -> str: if not all_skills: return _json({ "success": True, "skills": [], "categories": [], - "message": "No skills found in skills/ directory.", - }) + "message": "No skills found in skills/ directory."}) if category: all_skills = [s for s in all_skills if s.get("category") == category] all_skills = _sort_skills(all_skills) @@ -385,8 +321,7 @@ def skills_list(category: str = None, task_id: str = None) -> str: "skills": all_skills, "categories": categories, "count": len(all_skills), - "hint": "Use skill_view(name) to see full content, tags, and linked files", - }) + "hint": "Use skill_view(name) to see full content, tags, and linked files"}) except Exception as e: return tool_error(str(e), success=False) @@ -418,10 +353,9 @@ def _resolve_plugin_skill(name, file_path, task_id, preprocess): logger.debug("Failed pruning inactive memory-provider skills: %s", exc) plugin_skill_md = pm.find_plugin_skill(name) - # Memory provider plugins load through plugins.memory rather than the general - # PluginManager. If a memory provider shim also registers skills, load the - # namespaced provider once so its collector can forward those skills into the - # plugin skill registry before declaring the qualified skill missing. + # Memory providers load through plugins.memory, not the general PluginManager: + # load the namespaced provider once so its collector can forward its skills + # into the registry before declaring the qualified skill missing. if plugin_skill_md is None: try: if namespace == active_memory_provider: @@ -460,33 +394,24 @@ def _under_any(path: Path, dirs) -> bool: resolved = path.resolve() except Exception: resolved = path - for d in dirs: - try: - resolved.relative_to(d) - return True - except ValueError: - continue - return False + return any(resolved.is_relative_to(d) for d in dirs) def _collect_skill_candidates(name, local_category_name, all_dirs): """ALL (skill_dir, skill_md) candidates across every dir using every lookup strategy (direct path, recursive by parent dir / frontmatter name, legacy flat - .md), deduped by resolved path. - - Collision detection is the point: silent shadowing of a local skill by a - same-named external skill is a real bug class (`/skills` shows one, agent - loaded the other), so the caller refuses when more than one matches.""" + .md), deduped by resolved path. Collision detection is the point: silent + shadowing of a local skill by a same-named external one is a real bug class, + so the caller refuses when more than one matches.""" from agent.skill_utils import iter_skill_index_files candidates: List[Tuple[Optional[Path], Path]] = [] seen_md: set = set() def _record(sd: Optional[Path], smd: Path) -> None: - try: + key = smd + with suppress(Exception): key = smd.resolve() - except Exception: - key = smd if key not in seen_md: seen_md.add(key) candidates.append((sd, smd)) @@ -501,67 +426,56 @@ def _collect_skill_candidates(name, local_category_name, all_dirs): for search_dir in all_dirs: _record_direct(search_dir / name) - # Categorized form for plugin-namespace fall-through ("myplugin:explore" - # with no such plugin also tries the on-disk path "myplugin/explore"). - if local_category_name: + if local_category_name: # "myplugin:explore" with no such plugin → "myplugin/explore" _record_direct(search_dir / local_category_name) - # Recursive by directory name (nested skills called by bare name) plus - # frontmatter `name:` — skills_list() exposes the frontmatter name, so - # skill_view(name) must accept it even when the directory is an alias. + # Recursive by directory name plus frontmatter `name:` — skills_list() + # exposes the frontmatter name, so skill_view(name) must accept it too. for found_skill_md in iter_skill_index_files(search_dir, "SKILL.md"): if found_skill_md.parent.name == name: _record(found_skill_md.parent, found_skill_md) continue - try: + fm: dict = {} + with suppress(Exception): fm, _ = _parse_frontmatter(_read_skill_text(found_skill_md)) - except Exception: - fm = {} if fm.get("name") == name: _record(found_skill_md.parent, found_skill_md) - # Legacy flat .md anywhere under the dir. Skill support docs - # (references/templates/assets/scripts) are excluded: they load via - # file_path and must not shadow real skills sharing the basename. + # Legacy flat .md anywhere under the dir; support docs are excluded + # (they load via file_path and must not shadow real skills sharing the basename). for found_md in search_dir.rglob(f"{name}.md"): if found_md.name != "SKILL.md" and not _is_skill_support_path(found_md): _record(None, found_md) return candidates -_TEMPLATE_GLOBS = ["*.md", "*.py", "*.yaml", "*.yml", "*.json", "*.tex", "*.sh"] -_SCRIPT_GLOBS = ["*.py", "*.sh", "*.bash", "*.js", "*.ts", "*.rb"] +# (support dir, globs, recursive, files only) — order is the linked_files key order. +_LINKED_FILE_SPECS = ( + ("references", ["*.md"], False, False), + ("templates", ["*.md", "*.py", "*.yaml", "*.yml", "*.json", "*.tex", "*.sh"], True, False), + ("assets", ["*"], True, True), + ("scripts", ["*.py", "*.sh", "*.bash", "*.js", "*.ts", "*.rb"], False, False), +) def _skill_linked_files(skill_dir: Optional[Path]) -> dict: """references/templates/assets/scripts of a directory skill (empty groups dropped).""" - if not skill_dir: - return {} - - def _rel(paths): - return [str(f.relative_to(skill_dir)) for f in paths] - def _multi(sub: Path, globs, rglob: bool): - out: list = [] - for ext in globs: - out.extend(_rel(sub.rglob(ext) if rglob else sub.glob(ext))) - return out - - refs, tmpl, assets, scripts = (skill_dir / n for n in ("references", "templates", "assets", "scripts")) - files = { - "references": _rel(refs.glob("*.md")) if refs.exists() else [], - "templates": _multi(tmpl, _TEMPLATE_GLOBS, True) if tmpl.exists() else [], - "assets": _rel(f for f in assets.rglob("*") if f.is_file()) if assets.exists() else [], - "scripts": _multi(scripts, _SCRIPT_GLOBS, False) if scripts.exists() else [], - } - return {k: v for k, v in files.items() if v} + files: dict = {} + for sub, globs, recursive, files_only in _LINKED_FILE_SPECS if skill_dir else (): + base = skill_dir / sub + if not base.exists(): + continue + found = [ + str(f.relative_to(skill_dir)) + for g in globs for f in (base.rglob(g) if recursive else base.glob(g)) + if not files_only or f.is_file()] + if found: + files[sub] = found + return files def _org_provenance_header(skill_dir: Path, active_skills_dir: Path): """(org_provenance dict, header text) for an org-mirror skill, else (None, ""). - - An org-shared skill announces its provenance IN the returned content — the - moment the model consumes it — not only in the listing. The commit author is - token-verified at push time by the sync plane (author_mismatch guard), so the - header is trustworthy, not client-claimed. Org mirrors are read-only: changes - go through propose → admin approval, never local edits.""" + Provenance is announced IN the content the model consumes; the author is + token-verified at push time by the sync plane, so the header is trustworthy.""" from agent.skill_utils import ORG_PROVENANCE_FILE, is_org_mirror_path, org_id_of_path if not is_org_mirror_path(skill_dir, active_skills_dir): @@ -569,28 +483,20 @@ def _org_provenance_header(skill_dir: Path, active_skills_dir: Path): prov_org = org_id_of_path(skill_dir, active_skills_dir) author = ts = "" if prov_org: - try: + with suppress(Exception): prov = json.loads(_read_skill_text(active_skills_dir / "_org" / prov_org / ORG_PROVENANCE_FILE)) author = str(prov.get("author_device") or prov.get("author_user_id") or "") ts = str(prov.get("ts") or "") - except Exception: - pass header = ( "> [!NOTE] ORG-SHARED SKILL — provenance\n" - f"> This skill is shared by your organisation (org " - f"`{prov_org}`" + f"> This skill is shared by your organisation (org `{prov_org}`" + (f", last updated by `{author}`" if author else "") + (f", as of {ts}" if ts else "") + "). It was reviewed and approved for the whole\n" - "> team — treat it as third-party instructions rather " - "than your own notes.\n" - "> You MAY improve it in place like any other skill. " - "Your edits are kept locally\n" - "> and are never overwritten by org updates; share " - "them back with\n" - "> `hermes sync propose` (or automatically, if your " - "org enables it).\n\n" - ) + "> team — treat it as third-party instructions rather than your own notes.\n" + "> You MAY improve it in place like any other skill. Your edits are kept locally\n" + "> and are never overwritten by org updates; share them back with\n" + "> `hermes sync propose` (or automatically, if your org enables it).\n\n") return {"org_id": prov_org, "shared_by": author or None, "as_of": ts or None}, header @@ -598,29 +504,23 @@ def _skill_readiness(frontmatter: Dict[str, Any], skill_name: str) -> Tuple[dict """Resolve required env vars / credential files (prompting for secrets where the surface allows) and register what's available for sandboxes. - Returns ``(fields, extras)``: ``fields`` are the readiness keys placed before - ``_source_path`` in the skill_view result; ``extras`` (setup_help, - gateway_setup_hint, setup_note) are appended after it — result key order is - part of the tool output.""" + Returns ``(fields, extras)``: ``fields`` go before ``_source_path`` in the + skill_view result, ``extras`` (setup_help, gateway_setup_hint, setup_note) + after it — result key order is part of the tool output.""" legacy_env_vars, _ = _collect_prerequisite_values(frontmatter) required_env_vars = _get_required_environment_variables(frontmatter, legacy_env_vars) backend = _get_terminal_backend_name() env_snapshot = load_env() missing_required_env_vars = [ e for e in required_env_vars - if not e.get("optional") and not _is_env_var_persisted(e["name"], env_snapshot) - ] + if not e.get("optional") and not _is_env_var_persisted(e["name"], env_snapshot)] capture_result = _capture_required_environment_variables(skill_name, missing_required_env_vars) if missing_required_env_vars: env_snapshot = load_env() - remaining = _remaining_required_environment_names( - required_env_vars, capture_result, env_snapshot=env_snapshot - ) + remaining = _remaining_required_environment_names(required_env_vars, capture_result, env_snapshot=env_snapshot) setup_needed = bool(remaining) - # Register available env vars so they pass through to sandboxed execution - # (execute_code, terminal). Only vars actually set get registered — missing - # ones are reported as setup_needed. + # Only vars actually set pass through to sandboxed execution (execute_code, terminal). available_env_names = [e["name"] for e in required_env_vars if e["name"] not in remaining] if available_env_names: try: @@ -630,8 +530,8 @@ def _skill_readiness(frontmatter: Dict[str, Any], skill_name: str) -> Tuple[dict except Exception: logger.debug("Could not register env passthrough for skill %s", skill_name, exc_info=True) - # Credential files for mounting into remote sandboxes (Modal, Docker): - # existing host files are registered, missing ones flag setup_needed. + # Credential files for mounting into remote sandboxes: existing host files are + # registered, missing ones flag setup_needed. required_cred_files_raw = frontmatter.get("required_credential_files", []) missing_cred_files: list = [] if isinstance(required_cred_files_raw, list) and required_cred_files_raw: @@ -654,8 +554,7 @@ def _skill_readiness(frontmatter: Dict[str, Any], skill_name: str) -> Tuple[dict "setup_skipped": capture_result["setup_skipped"], "readiness_status": ( SkillReadinessStatus.SETUP_NEEDED if setup_needed else SkillReadinessStatus.AVAILABLE - ).value, - } + ).value} extras: dict = {} setup_help = next((e["help"] for e in required_env_vars if e.get("help")), None) if setup_help: @@ -672,23 +571,93 @@ def _skill_readiness(frontmatter: Dict[str, Any], skill_name: str) -> Tuple[dict return fields, extras +@dataclass +class _Located: + """Outcome of the local skill lookup phase of ``skill_view``.""" + + project_dirs: list + all_dirs: list + active_skills_dir: Path + skill_dir: Optional[Path] = None + skill_md: Any = None # set whenever error is None + error: Optional[str] = None # JSON failure payload to return as-is + + +def _locate_skill(name: str, local_category_name: Optional[str]) -> _Located: + """Find the unique on-disk skill for *name*: collision refusal, project-tier + precedence, quarantine gate, not-found listing.""" + project_dirs, all_dirs, active_skills_dir = _skill_search_dirs() + loc = _Located(project_dirs, all_dirs, active_skills_dir) + if not all_dirs: + loc.error = _fail("Skills directory does not exist yet. It will be created on first install.") + return loc + + candidates = _collect_skill_candidates(name, local_category_name, all_dirs) + if len(candidates) > 1 and project_dirs: + # A project skill intentionally overrides a same-named local/external skill, + # so narrow to project candidates when any exist. Ambiguity WITHIN that tier still refuses. + project_candidates = [(sd, smd) for sd, smd in candidates if _under_any(smd, project_dirs)] + if project_candidates: + candidates = project_candidates + if len(candidates) > 1: + paths = [str(smd) for _, smd in candidates] + logger.warning("Skill name collision for '%s': %d candidates — %s", name, len(candidates), "; ".join(paths)) + loc.error = _fail( + f"Ambiguous skill name '{name}': {len(candidates)} skills match across your local skills dir " + "and external_dirs. Refusing to guess — load one explicitly by its categorized path.", + matches=paths, + hint="Pass the full relative path instead of the bare name (e.g., 'category/skill-name'), " + "or rename one of the colliding skills so each name is unique.") + return loc + if candidates: + loc.skill_dir, loc.skill_md = candidates[0] + + # Quarantine gate: a project-tier skill with a dangerous scan verdict must not + # load even by explicit name (same chokepoint the index and skills_list use). + if loc.skill_md is not None and project_dirs: + from agent.skill_utils import is_quarantined_project_skill + + if _under_any(loc.skill_md, project_dirs) and is_quarantined_project_skill(loc.skill_md): + loc.error = _fail( + f"Project skill '{name}' is quarantined: the security scan flagged its content as " + "dangerous. It will not load until the repo's skill content changes and passes a re-scan.", + hint="Inspect the skill in the repo checkout, or untrust the repo with `hermes skills untrust`.", + ) + return loc + + if not loc.skill_md or not loc.skill_md.exists(): + available = [s["name"] for s in _sort_skills(_find_all_skills())[:20]] + loc.error = _fail(f"Skill '{name}' not found.", available_skills=available, hint="Use skills_list to see all available skills") + return loc + + +def _log_security_warnings(name: str, skill_md: Path, content: str, loc: _Located) -> None: + """Warn when loaded from outside the trusted dirs (project + local + external) + and/or when common prompt-injection patterns appear. Never blocks.""" + trusted_dirs = [loc.active_skills_dir.resolve()] + with suppress(Exception): + trusted_dirs.extend(d.resolve() for d in loc.all_dirs) + warnings = [] + if not _under_any(skill_md, trusted_dirs): + warnings.append(f"skill file is outside the trusted skills directory (~/.hermes/skills/): {skill_md}") + content_lower = content.lower() + if any(p in content_lower for p in _INJECTION_PATTERNS): + warnings.append("skill content contains patterns that may indicate prompt injection") + if warnings: + logger.warning("Skill security warning for '%s': %s", name, "; ".join(warnings)) + + def skill_view( - name: str, - file_path: str = None, - task_id: str = None, - preprocess: bool = True, -) -> str: + name: str, file_path: str = None, task_id: str = None, preprocess: bool = True) -> str: """View a skill (SKILL.md) or a specific file within its directory, as JSON. ``name`` is a skill name or path ("axolotl", "03-fine-tuning/axolotl"); the qualified form "plugin:skill" resolves plugin-provided skills. ``preprocess`` - applies the configured SKILL.md template / inline shell rendering to main - content; internal slash/preload callers disable it because they render the - skill message themselves.""" + applies the configured SKILL.md template / inline shell rendering; internal + slash/preload callers disable it because they render the skill message themselves.""" try: # Validate before the ':' dispatch so a Windows drive path (C:\skills\foo) - # can't be reinterpreted as a plugin namespace, and so a traversal / - # absolute name never reaches the search-dir join. + # can't be reinterpreted as a plugin namespace. lookup_error = _skill_lookup_path_error(name) if lookup_error: return _fail(lookup_error, hint=_LOOKUP_HINT) @@ -698,81 +667,27 @@ def skill_view( served, local_category_name = _resolve_plugin_skill(name, file_path, task_id, preprocess) if served is not None: return served - # The categorized fall-through form (namespace/bare) joins onto each - # search dir too; re-validate it since `bare` is not namespace-checked. + # The fall-through form (namespace/bare) joins onto each search dir too; + # re-validate it since `bare` is not namespace-checked. if local_category_name: lookup_error = _skill_lookup_path_error(local_category_name) if lookup_error: return _fail(lookup_error, hint=_LOOKUP_HINT) - project_dirs, all_dirs, active_skills_dir = _skill_search_dirs() - if not all_dirs: - return _fail("Skills directory does not exist yet. It will be created on first install.") - - skill_dir = skill_md = None - candidates = _collect_skill_candidates(name, local_category_name, all_dirs) - if len(candidates) > 1 and project_dirs: - # Cross-tier collision resolution: a project skill intentionally - # overrides a same-named local/external skill, so narrow to project - # candidates when any exist. Ambiguity WITHIN that tier still refuses. - project_candidates = [(sd, smd) for sd, smd in candidates if _under_any(smd, project_dirs)] - if project_candidates: - candidates = project_candidates - if len(candidates) > 1: - paths = [str(smd) for _, smd in candidates] - logger.warning("Skill name collision for '%s': %d candidates — %s", name, len(candidates), "; ".join(paths)) - return _fail( - f"Ambiguous skill name '{name}': {len(candidates)} skills match across your local skills dir " - "and external_dirs. Refusing to guess — load one explicitly by its categorized path.", - matches=paths, - hint="Pass the full relative path instead of the bare name (e.g., 'category/skill-name'), " - "or rename one of the colliding skills so each name is unique.", - ) - if candidates: - skill_dir, skill_md = candidates[0] - - # Quarantine gate: a project-tier skill with a dangerous scan verdict must - # not load even by explicit name (same chokepoint the index and - # skills_list use — see agent.skill_utils.iter_project_skill_files). - if skill_md is not None and project_dirs: - from agent.skill_utils import is_quarantined_project_skill - - if _under_any(skill_md, project_dirs) and is_quarantined_project_skill(skill_md): - return _fail( - f"Project skill '{name}' is quarantined: the security scan flagged its content as " - "dangerous. It will not load until the repo's skill content changes and passes a re-scan.", - hint="Inspect the skill in the repo checkout, or untrust the repo with `hermes skills untrust`.", - ) - - if not skill_md or not skill_md.exists(): - available = [s["name"] for s in _sort_skills(_find_all_skills())[:20]] - return _fail(f"Skill '{name}' not found.", available_skills=available, hint="Use skills_list to see all available skills") + loc = _locate_skill(name, local_category_name) + if loc.error is not None: + return loc.error + skill_dir, skill_md, active_skills_dir = loc.skill_dir, loc.skill_md, loc.active_skills_dir try: # read once — reused for platform check and main content content = _read_skill_text(skill_md) except Exception as e: return _fail(f"Failed to read skill '{name}': {e}") + _log_security_warnings(name, skill_md, content, loc) - # Security warnings: loaded from outside the trusted dirs (project + - # local + external = all_dirs), and/or common prompt-injection patterns. - _trusted_dirs = [active_skills_dir.resolve()] - try: - _trusted_dirs.extend(d.resolve() for d in all_dirs) - except Exception: - pass - _warnings = [] - if not _under_any(skill_md, _trusted_dirs): - _warnings.append(f"skill file is outside the trusted skills directory (~/.hermes/skills/): {skill_md}") - _content_lower = content.lower() - if any(p in _content_lower for p in _INJECTION_PATTERNS): - _warnings.append("skill content contains patterns that may indicate prompt injection") - if _warnings: - logger.warning("Skill security warning for '%s': %s", name, "; ".join(_warnings)) - - try: + frontmatter: Dict[str, Any] = {} + with suppress(Exception): frontmatter, _ = _parse_frontmatter(content) - except Exception: - frontmatter = {} if not skill_matches_platform(frontmatter): return _fail(f"Skill '{name}' is not supported on this platform.", readiness_status=SkillReadinessStatus.UNSUPPORTED.value) @@ -784,8 +699,7 @@ def skill_view( return _serve_skill_file( skill_dir, file_path, name, hint="Use a relative path within the skill directory", - list_available=True, mark_read=True, - ) + list_available=True, mark_read=True) # tags/related_skills: metadata.hermes.* (agentskills.io) first, then top-level. metadata = frontmatter.get("metadata") @@ -829,8 +743,7 @@ def skill_view( **readiness, # Internal: absolute source path for the repeat-view dedup fingerprint. "_source_path": str(skill_md), - **readiness_extras, - } + **readiness_extras} _mark_background_review_read(skill_md) # agentskills.io optional fields if frontmatter.get("compatibility"): @@ -884,21 +797,16 @@ registry.register( schema=SKILLS_LIST_SCHEMA, handler=lambda args, **kw: skills_list(category=args.get("category"), task_id=kw.get("task_id")), check_fn=check_skills_requirements, - emoji="📚", -) + emoji="📚") def _skill_view_with_bump(args, **kw): - """Invoke skill_view, then bump view_count on success. Best-effort: a - telemetry failure never breaks the tool call. + """Invoke skill_view, then bump view_count/use on success (best-effort). - Repeat-view dedup mirrors read_file's unchanged-stub: when this session - already loaded the SAME skill file and it hasn't changed on disk, return a - short stub instead of the full content (production mining: ~286k tokens of - verbatim repeat skill_view content in one 400k-message window). The stub only - replaces content already fully present earlier in this conversation, so the - "skills must be loaded fully" rule holds — and the cache is cleared on - context compression so a post-compression re-view returns full content.""" + Repeat-view dedup mirrors read_file's unchanged-stub: when this session already + loaded the SAME unchanged skill file, return a short stub instead of the full + content. The stub only replaces content already present earlier in this + conversation, and the cache is cleared on context compression.""" name = args.get("name", "") task_id = kw.get("task_id") stub = _check_skill_view_dedup(task_id, name, args.get("file_path")) @@ -909,15 +817,13 @@ def _skill_view_with_bump(args, **kw): parsed = json.loads(result) if isinstance(parsed, dict) and parsed.get("success"): _record_skill_view(task_id, name, args.get("file_path"), parsed) - # Resolved name from the payload when present — qualified forms - # ("plugin:skill") return with the canonical name. + # Qualified forms ("plugin:skill") return with the canonical name. resolved = parsed.get("name") or name if resolved: from tools.skill_usage import bump_use, bump_view bump_view(str(resolved)) - # A skill_view call is the agent actively loading the skill to act - # on it — that counts as use, not just a browse. Curator's stale - # timer keys off last_used_at (see agent/curator.py). + # Viewing is the agent actively loading the skill to act on it — that + # counts as use; the curator's stale timer keys off last_used_at. bump_use(str(resolved), task_id=kw.get("task_id"), session_id=kw.get("session_id")) except Exception: pass @@ -930,5 +836,4 @@ registry.register( schema=SKILL_VIEW_SCHEMA, handler=_skill_view_with_bump, check_fn=check_skills_requirements, - emoji="📚", -) + emoji="📚") diff --git a/tools/skills_tool_dedup.py b/tools/skills_tool_dedup.py index dfd4a03e11..9a2aa86ac9 100644 --- a/tools/skills_tool_dedup.py +++ b/tools/skills_tool_dedup.py @@ -1,14 +1,10 @@ """skill_view repeat-view dedup registry. -Per-task cache of (skill name, file_path) -> (skill file mtime+size). On a -repeat view of an UNCHANGED skill file, ``tools.skills_tool`` returns a short -stub instead of re-sending the full content — the earlier tool result in this -conversation already carries it verbatim. Cleared on context compression via -``reset_skill_view_dedup()`` (wired next to read_file's reset_file_dedup) -because after compression the original content is summarized away. - -Every name is re-imported into ``tools.skills_tool``; the tracker state lives -here and only here. +Per-task cache of (skill name, file_path) -> (skill file mtime+size). A repeat +view of an UNCHANGED file returns a short stub — the earlier tool result already +carries the content verbatim. Cleared on context compression via +``reset_skill_view_dedup()`` because the original content is summarized away. +Every name is re-imported into ``tools.skills_tool``; state lives only here. """ import json @@ -24,8 +20,7 @@ _SKILL_VIEW_DEDUP_MESSAGE = ( "Skill content unchanged since it was loaded earlier in this " "conversation — refer to the earlier skill_view result; it is still " "current and complete. (Re-issued after context compression, this " - "returns the full content again.)" -) + "returns the full content again.)") def _skill_view_fingerprint(payload: dict) -> tuple | None: @@ -44,9 +39,8 @@ def _record_skill_view(task_id, name, file_path, payload: dict) -> None: """Record a served skill_view so an identical repeat can be deduped.""" if not task_id: return - # Never dedup setup-needed views: readiness depends on config/env state - # that can change without the skill file changing, and the model must - # see the refreshed setup status on a re-view. + # Never dedup setup-needed views: readiness depends on config/env state that + # changes without the file changing; the model must see the refreshed status. if payload.get("setup_needed") or payload.get("readiness_status") == "setup_needed": return fp = _skill_view_fingerprint(payload) @@ -64,16 +58,16 @@ def _record_skill_view(task_id, name, file_path, payload: dict) -> None: def _check_skill_view_dedup(task_id, name, file_path) -> str | None: - """Return a dedup stub when this exact skill file was already served - to this task and is unchanged on disk; None otherwise.""" + """Dedup stub when this exact skill file was already served to this task and + is unchanged on disk; None otherwise.""" if not task_id: return None with _skill_view_tracker_lock: cache = _skill_view_tracker.get(str(task_id)) if not cache: return None - # The record key uses the RESOLVED name; check both the raw arg and - # resolved forms so 'category/skill' and bare-name views coalesce. + # Record key is the RESOLVED name; match raw and resolved forms so + # 'category/skill' and bare-name views coalesce. for key, (src, mtime_ns, size) in list(cache.items()): rec_name, rec_fp = key if rec_fp != (file_path or ""): @@ -99,9 +93,7 @@ def _check_skill_view_dedup(task_id, name, file_path) -> str | None: def reset_skill_view_dedup(task_id: str | None = None) -> None: - """Clear the skill_view dedup cache (all tasks when task_id is None). Called on - context compression: the original content is summarized away, so a re-view - must return full content again.""" + """Clear the dedup cache (all tasks when task_id is None); called on context compression.""" with _skill_view_tracker_lock: if task_id is None: _skill_view_tracker.clear() diff --git a/tools/skills_tool_plugin.py b/tools/skills_tool_plugin.py index 73e619ec57..d19d87bea4 100644 --- a/tools/skills_tool_plugin.py +++ b/tools/skills_tool_plugin.py @@ -1,10 +1,9 @@ """Plugin-provided skill serving for ``skill_view`` (``plugin:skill`` names) plus the JSON / file-serving helpers shared with the local-skill path. -Split out of ``tools.skills_tool``; every name is re-imported there. Helpers -that tests patch on the origin module (``_is_skill_disabled``, -``_parse_frontmatter``, ``skill_matches_platform``) are looked up lazily via -``tools.skills_tool`` at call time so those patches keep working. +Every name is re-imported by ``tools.skills_tool``. Helpers tests patch on the +origin module (``_is_skill_disabled``, ``_parse_frontmatter``, +``skill_matches_platform``) are looked up lazily via ``tools.skills_tool``. """ import json @@ -24,8 +23,7 @@ MAX_DESCRIPTION_LENGTH = 1024 _INJECTION_PATTERNS: list = [ "ignore previous instructions", "ignore all previous", "you are now", "disregard your", "forget your instructions", "new instructions:", - "system prompt:", "", "]]>", -] + "system prompt:", "", "]]>"] _SUPPORT_DIRS = ("references", "templates", "assets", "scripts") _SKILL_FILE_EXTS = {".md", ".py", ".yaml", ".yml", ".json", ".tex", ".sh"} @@ -39,10 +37,8 @@ def _fail(error: str, **extra) -> str: def _read_skill_text(path: Path) -> str: - """utf-8-sig + errors="replace": SKILL.md files are user-authored and may carry - a Notepad BOM or stray non-UTF-8 bytes. Pinning UTF-8 with replacement keeps - skill_view deterministic across platforms — falling back to the machine locale - (cp1252/GBK) would render the same skill differently per host (PR #51701).""" + """utf-8-sig + errors="replace": user-authored SKILL.md may carry a Notepad BOM + or stray bytes; pinning UTF-8 keeps skill_view deterministic across host locales.""" return path.read_text(encoding="utf-8-sig", errors="replace") @@ -76,14 +72,12 @@ def _serve_skill_file( hint: str | None = None, list_available: bool = False, read_error_prefix: bool = False, - mark_read: bool = False, -) -> str: + mark_read: bool = False) -> str: """Serve one linked file from a skill directory as a skill_view JSON result. - ``hint`` is attached to traversal/containment errors (local skills only); - ``list_available`` adds the available-files listing on not-found (local); - ``read_error_prefix`` wraps non-decode read errors (plugin) instead of letting - them propagate to the caller's generic handler (local).""" + ``hint`` decorates traversal/containment errors and ``list_available`` adds the + available-files listing on not-found (local skills); ``read_error_prefix`` wraps + non-decode read errors (plugin) instead of propagating to the caller's handler.""" from tools.path_security import has_traversal_component, validate_within_dir extra = {"hint": hint} if hint else {} @@ -93,16 +87,14 @@ def _serve_skill_file( path_error = validate_within_dir(target, skill_root) if path_error: return _fail(path_error, **extra) - # is_file(), not exists(): a directory (e.g. requesting 'references' bare) - # must take the not-found branch, not surface a raw [Errno 21] from read_text(). + # is_file(), not exists(): a bare directory must take the not-found branch. if not target.is_file(): not_found = f"File '{file_path}' not found in skill '{label}'." if list_available: return _fail( not_found, available_files=_available_skill_files(skill_root), - hint="Use one of the available file paths listed above", - ) + hint="Use one of the available file paths listed above") return _fail(not_found) try: content = _read_skill_text(target) @@ -110,8 +102,7 @@ def _serve_skill_file( return _json({ "success": True, "name": label, "file": file_path, "content": f"[Binary file: {target.name}, size: {target.stat().st_size} bytes]", - "is_binary": True, - }) + "is_binary": True}) except Exception as exc: if not read_error_prefix: raise @@ -122,8 +113,7 @@ def _serve_skill_file( "success": True, "name": label, "file": file_path, "content": content, "file_type": target.suffix, # Internal: absolute source path for the repeat-view dedup fingerprint. - "_source_path": str(target), - }) + "_source_path": str(target)}) def _mark_background_review_read(path: Path) -> None: @@ -151,8 +141,7 @@ def _serve_plugin_skill( file_path: str | None = None, *, preprocess: bool = True, - session_id: str | None = None, -) -> str: + session_id: str | None = None) -> str: """Read a plugin-provided skill, apply guards, return JSON.""" from hermes_cli.plugins import _get_disabled_plugins, get_plugin_manager from tools import skills_tool as _st @@ -174,26 +163,21 @@ def _serve_plugin_skill( if not _st.skill_matches_platform(parsed_frontmatter): return _fail( f"Skill '{qualified_name}' is not supported on this platform.", - readiness_status=SkillReadinessStatus.UNSUPPORTED.value, - ) + readiness_status=SkillReadinessStatus.UNSUPPORTED.value) if file_path: return _serve_skill_file(skill_md.parent, file_path, qualified_name, read_error_prefix=True) - # Injection scan — log but still serve (matches local-skill behaviour) if any(p in content.lower() for p in _INJECTION_PATTERNS): logger.warning( "Plugin skill '%s:%s' contains patterns that may indicate prompt injection", - namespace, bare, - ) - # Bundle context banner — tells the agent about sibling skills - try: + namespace, bare) + try: # bundle-context banner: sibling skills of the same plugin siblings = [s for s in get_plugin_manager().list_plugin_skills(namespace) if s != bare] banner = f"[Bundle context: This skill is part of the '{namespace}' plugin." if siblings: banner += ( f"\nSibling skills: {', '.join(siblings)}.\n" - f"Use qualified form to invoke siblings (e.g. {namespace}:{siblings[0]})." - ) + f"Use qualified form to invoke siblings (e.g. {namespace}:{siblings[0]}).") banner += "]\n\n" except Exception: banner = "" @@ -201,16 +185,14 @@ def _serve_plugin_skill( if preprocess: rendered_content = _preprocess_skill( content, skill_md.parent, session_id, - "Could not preprocess plugin skill %s:%s", namespace, bare, - ) + "Could not preprocess plugin skill %s:%s", namespace, bare) return _json({ "success": True, "name": qualified_name, "content": f"{banner}{rendered_content}" if banner else rendered_content, "description": _truncate_description(str(parsed_frontmatter.get("description", ""))), "linked_files": _plugin_skill_linked_files(skill_md.parent), - "readiness_status": SkillReadinessStatus.AVAILABLE.value, - }) + "readiness_status": SkillReadinessStatus.AVAILABLE.value}) def _plugin_skill_linked_files(skill_root: Path) -> Dict[str, List[str]] | None: @@ -224,8 +206,7 @@ def _plugin_skill_linked_files(skill_root: Path) -> Dict[str, List[str]] | None: files = [ str(path.relative_to(skill_root)) for path in sorted(base.rglob("*")) - if path.is_file() and validate_within_dir(path, skill_root) is None - ] + if path.is_file() and validate_within_dir(path, skill_root) is None] if files: linked[category] = files return linked or None diff --git a/tools/skills_tool_setup.py b/tools/skills_tool_setup.py index d00f19eaed..4c9b9189e1 100644 --- a/tools/skills_tool_setup.py +++ b/tools/skills_tool_setup.py @@ -1,10 +1,8 @@ """Skill readiness: required env vars, secret capture, and setup notes. -Split out of ``tools.skills_tool``; every name is re-imported there so -``from tools.skills_tool import X`` / ``patch("tools.skills_tool.X")`` keep -working. Module state (``_secret_capture_callback``, ``load_env``) stays in -``tools.skills_tool`` and is read lazily at call time so test patches on the -origin module are honored. +Split out of ``tools.skills_tool``; every name is re-imported there. Module +state (``_secret_capture_callback``, ``load_env``) stays in ``tools.skills_tool`` +and is read lazily at call time so test patches on the origin module are honored. """ import logging @@ -56,8 +54,7 @@ def _collect_prerequisite_values(frontmatter: Dict[str, Any]) -> Tuple[List[str] return [], [] return ( _normalize_prerequisite_values(prereqs.get("env_vars")), - _normalize_prerequisite_values(prereqs.get("commands")), - ) + _normalize_prerequisite_values(prereqs.get("commands"))) def _as_dict_list(raw: Any) -> list: @@ -86,8 +83,7 @@ def _normalize_setup_metadata(frontmatter: Dict[str, Any]) -> Dict[str, Any]: entry: Dict[str, Any] = { "env_var": env_var, "prompt": str(item.get("prompt") or f"Enter value for {env_var}").strip(), - "secret": bool(item.get("secret", True)), - } + "secret": bool(item.get("secret", True))} provider_url = str(item.get("provider_url") or item.get("url") or "").strip() if provider_url: entry["provider_url"] = provider_url @@ -96,9 +92,7 @@ def _normalize_setup_metadata(frontmatter: Dict[str, Any]) -> Dict[str, Any]: def _get_required_environment_variables( - frontmatter: Dict[str, Any], - legacy_env_vars: List[str] | None = None, -) -> List[Dict[str, Any]]: + frontmatter: Dict[str, Any], legacy_env_vars: List[str] | None = None) -> List[Dict[str, Any]]: """Merge required_environment_variables, setup.collect_secrets and legacy prerequisites.env_vars into one deduped, validated list (first entry wins).""" setup = _normalize_setup_metadata(frontmatter) @@ -111,11 +105,9 @@ def _get_required_environment_variables( return normalized: Dict[str, Any] = { "name": env_name, - "prompt": str(entry.get("prompt") or f"Enter value for {env_name}").strip(), - } + "prompt": str(entry.get("prompt") or f"Enter value for {env_name}").strip()} help_text = _clean_str( - entry.get("help") or entry.get("provider_url") or entry.get("url") or setup.get("help") - ) + entry.get("help") or entry.get("provider_url") or entry.get("url") or setup.get("help")) if help_text: normalized["help"] = help_text required_for = _clean_str(entry.get("required_for")) @@ -135,8 +127,7 @@ def _get_required_environment_variables( _append_required({ "name": item.get("env_var"), "prompt": item.get("prompt"), - "help": item.get("provider_url") or setup.get("help"), - }) + "help": item.get("provider_url") or setup.get("help")}) if legacy_env_vars is None: legacy_env_vars, _ = _collect_prerequisite_values(frontmatter) for env_var in legacy_env_vars: @@ -145,28 +136,21 @@ def _get_required_environment_variables( def _capture_result(missing_names, setup_skipped=False, gateway_setup_hint=None): - return { - "missing_names": missing_names, - "setup_skipped": setup_skipped, - "gateway_setup_hint": gateway_setup_hint, - } + return {"missing_names": missing_names, "setup_skipped": setup_skipped, "gateway_setup_hint": gateway_setup_hint} def _capture_required_environment_variables( - skill_name: str, - missing_entries: List[Dict[str, Any]], -) -> Dict[str, Any]: + skill_name: str, missing_entries: List[Dict[str, Any]]) -> Dict[str, Any]: """Prompt for missing secrets via the registered capture callback (if any).""" from tools import skills_tool as _st if not missing_entries: return _capture_result([]) missing_names = [entry["name"] for entry in missing_entries] - # Most gateway surfaces (messaging platforms) can't prompt for a secret, so - # they short-circuit to the "unsupported" hint. Interactive gateway surfaces - # (desktop app / TUI) set HERMES_INTERACTIVE — the same flag tools/approval.py - # uses — and register a callback routing to a secure secret.request overlay, - # so they fall through and actually prompt. + # Messaging-platform gateway surfaces can't prompt for a secret, so they get + # the "unsupported" hint. Interactive gateway surfaces (desktop app / TUI) set + # HERMES_INTERACTIVE (same flag tools/approval.py uses) and register a callback + # routing to a secure secret.request overlay, so they fall through and prompt. if _is_gateway_surface() and not env_var_enabled("HERMES_INTERACTIVE"): return _capture_result(missing_names, gateway_setup_hint=_gateway_setup_hint()) callback = _st._secret_capture_callback @@ -223,16 +207,14 @@ def _remaining_required_environment_names( required_env_vars: List[Dict[str, Any]], capture_result: Dict[str, Any], *, - env_snapshot: Dict[str, str] | None = None, -) -> List[str]: + env_snapshot: Dict[str, str] | None = None) -> List[str]: missing_names = set(capture_result["missing_names"]) env_snapshot = _env_snapshot_or_load(env_snapshot) return [ e["name"] for e in required_env_vars if not e.get("optional") - and (e["name"] in missing_names or not _is_env_var_persisted(e["name"], env_snapshot)) - ] + and (e["name"] in missing_names or not _is_env_var_persisted(e["name"], env_snapshot))] def _gateway_setup_hint() -> str: @@ -245,9 +227,7 @@ def _gateway_setup_hint() -> str: def _build_setup_note( - readiness_status: SkillReadinessStatus, - missing: List[str], - setup_help: str | None = None, + readiness_status: SkillReadinessStatus, missing: List[str], setup_help: str | None = None ) -> str | None: if readiness_status != SkillReadinessStatus.SETUP_NEEDED: return None