ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 11 Packages Projects Releases Wiki Activity

Merge branch 'simp/r3-34-D' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:41:36 -07:00
parent 0657c03fa4 68d7766cbf
commit e4dbad91ce
7 changed files with 379 additions and 598 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/skills_sync.py
+46 -100
View File
@@ -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__":
tools/skills_sync_bundled_ops.py
+23 -42
View File
@@ -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)}."}
tools/skills_sync_optional.py
+24 -28
View File
@@ -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:
tools/skills_tool.py
+233 -328
View File
@@ -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.<attr>`` — 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
<name>.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."""
<name>.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 <name>.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 <name>.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="📚")
tools/skills_tool_dedup.py
+13 -21
View File
@@ -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()
tools/skills_tool_plugin.py
+22 -41
View File
@@ -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>", "]]>",
]
"system prompt:", "<system>", "]]>"]
_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
tools/skills_tool_setup.py
+18 -38
View File
@@ -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