207f8fedfd
# Conflicts: # hermes_cli/local_runtime/binaries.py # hermes_cli/plugins_cmd.py # hermes_constants.py # tests/test_hermes_constants.py # tests/tools/test_clipboard.py # tests/tools/test_voice_wsl_pipewire.py # tools/computer_use/cua_backend.py # tools/voice_mode.py
832 lines
35 KiB
Python
832 lines
35 KiB
Python
"""Lightweight skill metadata utilities shared by prompt_builder and skills_tool.
|
||
Import-light by design: no tool registry, CLI config, or provider resolution."""
|
||
|
||
import ast
|
||
import logging
|
||
import os
|
||
import re
|
||
import sys
|
||
from pathlib import Path, PurePath
|
||
from typing import Any, Callable, Dict, List, Optional, Set, Tuple
|
||
|
||
from hermes_constants import (
|
||
get_config_path,
|
||
get_skills_dir,
|
||
get_subprocess_home,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
PLATFORM_MAP = {"macos": "darwin", "linux": "linux", "windows": "win32"}
|
||
|
||
EXCLUDED_SKILL_DIRS = frozenset((
|
||
".git", ".github", ".hub", ".archive", ".curator_backups", ".locks",
|
||
".venv", "venv", "node_modules", "site-packages", "__pycache__",
|
||
".tox", ".nox", ".pytest_cache", ".mypy_cache", ".ruff_cache",
|
||
))
|
||
|
||
# Progressive-disclosure support dirs inside a skill package: loaded explicitly
|
||
# via skill_view(skill, file_path=...), never scanned as standalone skills.
|
||
SKILL_SUPPORT_DIRS = frozenset(("references", "templates", "assets", "scripts"))
|
||
|
||
# Org mirrors live under skills/_org/<org_id>/ and are TOKEN-GATED: the sync
|
||
# client writes the marker after verifying the token; no marker => no org skills
|
||
# load. The marker persists offline so already-pulled org skills keep working.
|
||
ORG_MIRROR_DIR_NAME = "_org"
|
||
ORG_ACTIVE_MARKER = ".active_org"
|
||
ORG_PROVENANCE_FILE = ".org-provenance.json"
|
||
ORG_BASELINE_FILE = ".org-baseline.json" # upstream fingerprint; detects local edits
|
||
|
||
|
||
def read_active_org_id(skills_dir: Path) -> Optional[str]:
|
||
"""The org id whose mirror may resolve, or None (no org skills load)."""
|
||
marker = skills_dir / ORG_MIRROR_DIR_NAME / ORG_ACTIVE_MARKER
|
||
try:
|
||
return (marker.read_text(encoding="utf-8-sig").strip() or None) if marker.exists() else None
|
||
except OSError:
|
||
return None
|
||
|
||
|
||
def _org_rel_parts(path, skills_dir: Path) -> Tuple[str, ...]:
|
||
"""Path parts of *path* relative to *skills_dir* if it is under ``_org/``, else ``()``."""
|
||
try:
|
||
parts = Path(path).resolve().relative_to(Path(skills_dir).resolve()).parts
|
||
except (OSError, ValueError):
|
||
return ()
|
||
return parts if parts and parts[0] == ORG_MIRROR_DIR_NAME else ()
|
||
|
||
|
||
def is_org_mirror_path(path, skills_dir: Path) -> bool:
|
||
"""True when *path* is inside the org mirror (``_org/``)."""
|
||
return bool(_org_rel_parts(path, skills_dir))
|
||
|
||
|
||
def org_id_of_path(path, skills_dir: Path) -> Optional[str]:
|
||
"""The ``<org_id>`` segment for a path under ``_org/<org_id>/...``."""
|
||
parts = _org_rel_parts(path, skills_dir)
|
||
return parts[1] if len(parts) >= 2 else None
|
||
|
||
|
||
def is_excluded_skill_path(path, *, root: Optional[Path] = None) -> bool:
|
||
"""True if *path* should be skipped by skill scanners (VCS/dependency/cache
|
||
dirs + support packages). Apply to every SKILL.md from a direct ``rglob``."""
|
||
parts = PurePath(str(path)).parts
|
||
return any(part in EXCLUDED_SKILL_DIRS for part in parts) or is_skill_support_path(path, root=root)
|
||
|
||
|
||
def is_skill_support_path(path, *, root: Optional[Path] = None) -> bool:
|
||
"""True if *path* is under a support dir sitting directly inside a skill root
|
||
(``skills/scripts/foo`` stays discoverable: no ``SKILL.md`` above ``scripts``)."""
|
||
path_obj = path if isinstance(path, Path) else Path(str(path))
|
||
parts = path_obj.parts
|
||
base = root if root is not None and not path_obj.is_absolute() else Path()
|
||
# Only components before the leaf can be containing support directories.
|
||
return any(
|
||
part in SKILL_SUPPORT_DIRS and (base / Path(*parts[:idx]) / "SKILL.md").exists()
|
||
for idx, part in enumerate(parts[:-1])
|
||
if idx > 0
|
||
)
|
||
|
||
|
||
_yaml_load_fn = None
|
||
|
||
|
||
def yaml_load(content: str):
|
||
"""Parse YAML with the shared safe loader, imported lazily."""
|
||
global _yaml_load_fn
|
||
if _yaml_load_fn is None:
|
||
from hermes_yaml import safe_load
|
||
_yaml_load_fn = safe_load
|
||
return _yaml_load_fn(content)
|
||
|
||
|
||
def parse_frontmatter(content: str) -> Tuple[Dict[str, Any], str]:
|
||
"""Parse YAML frontmatter from markdown; returns (frontmatter_dict, body).
|
||
Malformed YAML falls back to key:value line splitting. A leading UTF-8 BOM
|
||
(Windows editors) is stripped first or it would defeat the ``---`` fence check."""
|
||
content = content.removeprefix("\ufeff")
|
||
end_match = re.search(r"\n---\s*\n", content[3:]) if content.startswith("---") else None
|
||
if not end_match:
|
||
return {}, content
|
||
yaml_content = content[3 : end_match.start() + 3]
|
||
body = content[end_match.end() + 3 :]
|
||
frontmatter: Dict[str, Any] = {}
|
||
try:
|
||
parsed = yaml_load(yaml_content)
|
||
if isinstance(parsed, dict):
|
||
frontmatter = parsed
|
||
except Exception:
|
||
for line in yaml_content.strip().split("\n"):
|
||
if ":" in line:
|
||
key, value = line.split(":", 1)
|
||
frontmatter[key.strip()] = value.strip()
|
||
return frontmatter, body
|
||
|
||
|
||
def skill_matches_platform_list(platforms: Any) -> bool:
|
||
"""Return True when *platforms* is compatible with the current OS."""
|
||
if not platforms:
|
||
return True
|
||
if not isinstance(platforms, list):
|
||
platforms = [platforms]
|
||
current = sys.platform
|
||
for platform in platforms:
|
||
normalized = str(platform).lower().strip()
|
||
mapped = PLATFORM_MAP.get(normalized, normalized)
|
||
if current.startswith(mapped):
|
||
return True
|
||
return False
|
||
|
||
|
||
def skill_matches_platform(frontmatter: Dict[str, Any]) -> bool:
|
||
"""True when the skill's ``platforms:`` list (absent = all) matches this OS."""
|
||
return skill_matches_platform_list(frontmatter.get("platforms"))
|
||
|
||
|
||
# An ``environments:`` tag is a *relevance* gate for offer surfaces (index,
|
||
# autocomplete, slash commands), not a compatibility gate: an explicit load
|
||
# (skill_view, --skills) always succeeds. Detection is cached per process.
|
||
|
||
_ENV_DETECT_CACHE: Dict[str, bool] = {}
|
||
|
||
|
||
def _detect_kanban() -> bool:
|
||
# Mirror tools/kanban_tools.py: a dispatcher-spawned worker (env vars, but
|
||
# only when this execution OWNS the task — delegate children / in-process
|
||
# cron see the worker's vars) or a profile opted into the kanban toolset.
|
||
if os.getenv("HERMES_KANBAN_TASK") or os.getenv("HERMES_KANBAN_BOARD"):
|
||
try:
|
||
from agent.delegation_context import is_dispatcher_owned_worker_context
|
||
owned = is_dispatcher_owned_worker_context()
|
||
except Exception:
|
||
owned = True
|
||
if owned:
|
||
return True
|
||
try:
|
||
from tools.kanban_tools import _profile_has_kanban_toolset
|
||
return bool(_profile_has_kanban_toolset())
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def _detect_docker() -> bool:
|
||
try:
|
||
from hermes_constants import is_container
|
||
return is_container()
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
_ENV_DETECTORS: Dict[str, Callable[[], bool]] = {
|
||
"kanban": _detect_kanban, "docker": _detect_docker,
|
||
"s6": lambda: os.path.isdir("/run/s6") or os.path.isdir("/package/admin/s6-overlay"), # s6-overlay is PID 1 in the image
|
||
}
|
||
|
||
|
||
def _detect_environment(env: str) -> bool:
|
||
"""True when the named runtime environment is active (unknown => True).
|
||
Cached per process EXCEPT ``kanban``: that verdict is context-dependent
|
||
(delegate children / in-process cron see the worker's vars), so a
|
||
process-wide cache would leak the first asker's answer to the others."""
|
||
if env != "kanban" and env in _ENV_DETECT_CACHE:
|
||
return _ENV_DETECT_CACHE[env]
|
||
detector = _ENV_DETECTORS.get(env)
|
||
result = detector() if detector else True
|
||
_ENV_DETECT_CACHE[env] = result
|
||
return result
|
||
|
||
|
||
def skill_matches_environment(frontmatter: Dict[str, Any]) -> bool:
|
||
"""True when ANY declared ``environments:`` tag is active (absent = all;
|
||
unknown tags fail open). Offer-time filter only."""
|
||
environments = frontmatter.get("environments")
|
||
if not environments:
|
||
return True
|
||
tags = [str(env).lower().strip() for env in (environments if isinstance(environments, list) else [environments])]
|
||
return any(_detect_environment(tag) for tag in tags if tag)
|
||
|
||
|
||
def skill_matches_apps(frontmatter: Dict[str, Any]) -> bool:
|
||
"""True when every app named in ``requires_apps:`` has a registered declaration this host satisfies.
|
||
|
||
Names resolve through ``hermes_platform.declaration`` (registered by whoever owns the server,
|
||
e.g. the plugin loader); the check is the same ``availability()`` the MCP check_fn uses. An
|
||
unknown name hides the skill (fail closed). Offer-time filter, like ``environments:``.
|
||
"""
|
||
names = frontmatter.get("requires_apps")
|
||
if not names:
|
||
return True
|
||
from hermes_platform import declaration
|
||
from hermes_platform.resolver.availability import availability
|
||
|
||
for name in names if isinstance(names, list) else [names]:
|
||
decl = declaration.lookup(str(name).strip())
|
||
if decl is None or decl.app is None:
|
||
return False
|
||
if not availability(decl).offerable:
|
||
return False
|
||
return True
|
||
|
||
|
||
_RAW_CONFIG_CACHE: Dict[Tuple[str, int, int, int, int], Dict[str, Any]] = {}
|
||
|
||
|
||
def _raw_config_cache_clear() -> None:
|
||
"""Test hook — drop the shared raw config cache."""
|
||
_RAW_CONFIG_CACHE.clear()
|
||
|
||
|
||
def _config_cache_key(config_path: Path) -> Optional[Tuple[str, int, int, int, int]]:
|
||
"""``(path, *file_signature)`` identity of config.yaml, or None when unreadable/absent."""
|
||
try:
|
||
from utils import file_signature
|
||
return (str(config_path), *file_signature(config_path.stat()))
|
||
except OSError:
|
||
return None
|
||
|
||
|
||
def _load_raw_config() -> Dict[str, Any]:
|
||
"""Read config.yaml with an mtime+size keyed cache (no hermes_cli.config import)."""
|
||
config_path = get_config_path()
|
||
if not config_path.exists():
|
||
return {}
|
||
cache_key = _config_cache_key(config_path)
|
||
cached = _RAW_CONFIG_CACHE.get(cache_key) if cache_key is not None else None
|
||
if cached is not None:
|
||
return cached
|
||
try:
|
||
parsed = yaml_load(config_path.read_text(encoding="utf-8-sig"))
|
||
except Exception as e:
|
||
logger.debug("Could not read skill config %s: %s", config_path, e)
|
||
return {}
|
||
if not isinstance(parsed, dict):
|
||
return {}
|
||
if cache_key is not None:
|
||
_RAW_CONFIG_CACHE.clear()
|
||
_RAW_CONFIG_CACHE[cache_key] = parsed
|
||
return parsed
|
||
|
||
|
||
def _skills_cfg() -> Optional[Dict[str, Any]]:
|
||
"""The ``skills:`` mapping from config.yaml, or None when absent/malformed."""
|
||
skills_cfg = _load_raw_config().get("skills")
|
||
return skills_cfg if isinstance(skills_cfg, dict) else None
|
||
|
||
|
||
def _skills_cfg_get(key: str) -> Any:
|
||
"""``skills.<key>`` from config.yaml, or None when the section is absent/malformed."""
|
||
skills_cfg = _skills_cfg()
|
||
return skills_cfg.get(key) if skills_cfg is not None else None
|
||
|
||
|
||
def _expand_path(entry: str) -> Path:
|
||
"""Expand ``~`` and ``${VAR}`` in a config path entry."""
|
||
return Path(os.path.expanduser(os.path.expandvars(entry)))
|
||
|
||
|
||
def _home_relative(p: Path) -> Path:
|
||
"""Anchor a relative config path at HERMES_HOME; absolute paths pass through."""
|
||
from hermes_constants import get_hermes_home
|
||
return p if p.is_absolute() else get_hermes_home() / p
|
||
|
||
|
||
# Never disableable: `hermes-agent` is the agent's own operating manual and the
|
||
# system prompt points at it unconditionally.
|
||
ESSENTIAL_SKILLS: frozenset = frozenset({"hermes-agent"})
|
||
|
||
|
||
def get_disabled_skill_names(platform: str | None = None) -> Set[str]:
|
||
"""Disabled skill names from config.yaml: global list ∪ platform list
|
||
(*platform* defaults to ``HERMES_PLATFORM`` / ``HERMES_SESSION_PLATFORM``)."""
|
||
skills_cfg = _skills_cfg()
|
||
if skills_cfg is None:
|
||
return set()
|
||
from gateway.session_context import get_session_env
|
||
resolved_platform = platform or os.getenv("HERMES_PLATFORM") or get_session_env("HERMES_SESSION_PLATFORM")
|
||
disabled = _normalize_string_set(skills_cfg.get("disabled"))
|
||
platform_disabled = (skills_cfg.get("platform_disabled") or {}).get(resolved_platform) if resolved_platform else None
|
||
if platform_disabled is not None:
|
||
disabled |= _normalize_string_set(platform_disabled)
|
||
return disabled - ESSENTIAL_SKILLS
|
||
|
||
|
||
def parse_config_string_list(value) -> List[str]:
|
||
"""Normalize a config value that may hold a JSON-array string into a list.
|
||
``hermes config set`` stores lists as quoted JSON/Python-literal strings;
|
||
treating one as a single name would silently filter nothing. A scalar
|
||
string still means one name.
|
||
|
||
See #13026, #86661.
|
||
"""
|
||
if isinstance(value, str):
|
||
if value.strip().startswith("["):
|
||
try:
|
||
parsed = ast.literal_eval(value.strip())
|
||
except (ValueError, SyntaxError):
|
||
parsed = None
|
||
if isinstance(parsed, list):
|
||
return [str(item) for item in parsed]
|
||
return [value]
|
||
return [str(item) for item in value] if isinstance(value, (list, tuple, set, frozenset)) else []
|
||
|
||
|
||
def _normalize_string_set(values) -> Set[str]:
|
||
return {name.strip() for name in parse_config_string_list(values) if name.strip()}
|
||
|
||
|
||
# config identity -> resolved external dirs. Called once per skill during
|
||
# banner / tool-registry scans; re-resolving each time dominated cold-start.
|
||
_EXTERNAL_DIRS_CACHE: Dict[Tuple[str, int, int, int, int], List[Path]] = {}
|
||
|
||
|
||
def _external_dirs_cache_clear() -> None:
|
||
"""Test hook — drop the in-process cache."""
|
||
_EXTERNAL_DIRS_CACHE.clear()
|
||
_raw_config_cache_clear()
|
||
|
||
|
||
def _config_str_list(raw) -> List[str]:
|
||
"""A scalar-or-list config entry as a list of stripped non-empty strings."""
|
||
if isinstance(raw, str):
|
||
raw = [raw]
|
||
if not isinstance(raw, list):
|
||
return []
|
||
return [e for e in (str(entry).strip() for entry in raw) if e]
|
||
|
||
|
||
def get_external_skills_dirs() -> List[Path]:
|
||
"""Validated, deduplicated ``skills.external_dirs`` (existing dirs only). Entries
|
||
are ``~``/``${VAR}`` expanded, relative to HERMES_HOME; the local skills dir is skipped."""
|
||
config_path = get_config_path()
|
||
if not config_path.exists():
|
||
return []
|
||
full_key = _config_cache_key(config_path)
|
||
cache_key = full_key
|
||
cached = _EXTERNAL_DIRS_CACHE.get(cache_key) if cache_key is not None else None
|
||
if cached is not None:
|
||
return list(cached) # copy so callers can't mutate the cache
|
||
skills_cfg = _skills_cfg()
|
||
if skills_cfg is None:
|
||
return []
|
||
local_skills = get_skills_dir().resolve()
|
||
result: List[Path] = []
|
||
for entry in _config_str_list(skills_cfg.get("external_dirs")):
|
||
p = _home_relative(_expand_path(entry)).resolve()
|
||
if p == local_skills or p in result:
|
||
continue
|
||
if p.is_dir():
|
||
result.append(p)
|
||
else:
|
||
logger.debug("External skills dir does not exist, skipping: %s", p)
|
||
if cache_key is not None:
|
||
_EXTERNAL_DIRS_CACHE[cache_key] = list(result)
|
||
return result
|
||
|
||
|
||
def get_skill_create_dir() -> Optional[Path]:
|
||
"""Configured ``skills.create_dir`` (need not exist yet), or None when unset;
|
||
relative to HERMES_HOME; a value equal to the local skills dir counts as unset."""
|
||
raw = _skills_cfg_get("create_dir")
|
||
entry = str(raw).strip() if raw and isinstance(raw, (str, os.PathLike)) else ""
|
||
if not entry:
|
||
return None
|
||
p = _home_relative(_expand_path(entry))
|
||
try:
|
||
resolved = p.resolve()
|
||
except OSError:
|
||
resolved = p
|
||
try:
|
||
if resolved == get_skills_dir().resolve():
|
||
return None
|
||
except OSError:
|
||
pass
|
||
return resolved
|
||
|
||
|
||
def display_skill_create_dir() -> str:
|
||
"""User-facing path where new skills are created (``~/`` shorthand when
|
||
possible); tool schema descriptions and prompts follow ``skills.create_dir``."""
|
||
from hermes_constants import display_hermes_home
|
||
create_dir = get_skill_create_dir()
|
||
if create_dir is None:
|
||
return f"{display_hermes_home()}/skills/"
|
||
if create_dir.is_relative_to(Path.home()):
|
||
return "~/" + create_dir.relative_to(Path.home()).as_posix() + "/"
|
||
return create_dir.as_posix() + "/"
|
||
|
||
|
||
def get_all_skills_dirs() -> List[Path]:
|
||
"""Skill dirs: local ``~/.hermes/skills/`` first, then create_dir, then external.
|
||
Trusted project dirs are NOT included (higher precedence; see get_project_skills_dirs)."""
|
||
dirs = [get_skills_dir()]
|
||
create_dir = get_skill_create_dir()
|
||
if create_dir is not None and create_dir.is_dir():
|
||
dirs.append(create_dir)
|
||
dirs.extend(d for d in get_external_skills_dirs() if d not in dirs)
|
||
return dirs
|
||
|
||
|
||
# Project-local skills (<root>/.hermes/skills, <root>/.agents/skills; root = nearest
|
||
# .git ancestor) are a prompt-injection vector if auto-sourced from any clone, so
|
||
# they load only when the root is in ``skills.trusted_project_dirs``; then they
|
||
# override same-named profile/bundled skills. cwd + trust list are session-fixed
|
||
# so the skills index stays byte-stable.
|
||
|
||
PROJECT_SKILLS_SUBDIRS = (os.path.join(".hermes", "skills"), os.path.join(".agents", "skills"))
|
||
|
||
_PROJECT_ROOT_MAX_DEPTH = 64 # walk-up bound for pathological cwds
|
||
|
||
|
||
def find_project_root(start: Optional[Path] = None) -> Optional[Path]:
|
||
"""Nearest ancestor containing ``.git`` (dir or worktree file), or None.
|
||
Without *start*, the surface's effective working directory wins over the process cwd — the same
|
||
ladder every other cwd consumer reads (``resolve_agent_cwd``: session-bound cwd, then the scope's
|
||
``TERMINAL_CWD``, then the process cwd). The session cwd comes first because a multi-session host
|
||
(TUI/desktop gateway) pins each session's workspace there while its terminal scope resolves a
|
||
placeholder ``terminal.cwd`` to ``$HOME``; reading only the scope made every project skill invisible
|
||
on those surfaces (#114359). ``TERMINAL_CWD`` is the per-surface workdir the terminal tool and cron
|
||
jobs use (a cron job sets it from its per-job ``workdir`` without chdir'ing the scheduler process),
|
||
which lets non-interactive surfaces inherit a prior interactive trust decision by project identity —
|
||
and a surface with no workdir in a trusted repo simply resolves no project and loads nothing (#48975).
|
||
"""
|
||
try:
|
||
if start is None:
|
||
from agent.runtime_cwd import resolve_agent_cwd
|
||
start = resolve_agent_cwd()
|
||
cur = Path(start).resolve()
|
||
except OSError:
|
||
return None
|
||
home = Path.home().resolve()
|
||
try:
|
||
for _ in range(_PROJECT_ROOT_MAX_DEPTH):
|
||
if (cur / ".git").exists():
|
||
# A dotfiles checkout AT home would make every session
|
||
# project-scoped; treat home itself as non-project.
|
||
return None if cur == home else cur
|
||
if cur.parent == cur:
|
||
return None
|
||
cur = cur.parent
|
||
except OSError:
|
||
pass
|
||
return None
|
||
|
||
|
||
def _project_trusted_dirs_from_config() -> Set[Path]:
|
||
"""Resolved set of trusted project roots from ``skills.trusted_project_dirs``."""
|
||
result: Set[Path] = set()
|
||
for entry in _config_str_list(_skills_cfg_get("trusted_project_dirs")):
|
||
try:
|
||
result.add(_expand_path(entry).resolve())
|
||
except OSError:
|
||
continue
|
||
return result
|
||
|
||
|
||
def is_project_root_trusted(root: Path) -> bool:
|
||
"""True when *root* is listed in ``skills.trusted_project_dirs``."""
|
||
try:
|
||
return Path(root).resolve() in _project_trusted_dirs_from_config()
|
||
except OSError:
|
||
return False
|
||
|
||
|
||
def _candidate_project_skills_dirs(root: Path) -> List[Path]:
|
||
"""Existing skill dirs under *root*, excluding the profile's own skills dir
|
||
(HERMES_HOME itself may live inside a git checkout)."""
|
||
local_skills = get_skills_dir().resolve()
|
||
dirs: List[Path] = []
|
||
for cand in (root / sub for sub in PROJECT_SKILLS_SUBDIRS):
|
||
try:
|
||
if cand.is_dir() and cand.resolve() != local_skills:
|
||
dirs.append(cand.resolve())
|
||
except OSError:
|
||
continue
|
||
return dirs
|
||
|
||
|
||
def _current_project_root(trusted: bool) -> Optional[Path]:
|
||
"""cwd's project root when discovery is on and its trust state == *trusted*."""
|
||
if _skills_cfg_get("project_discovery") is False:
|
||
return None
|
||
root = find_project_root()
|
||
return root if root is not None and is_project_root_trusted(root) == trusted else None
|
||
|
||
|
||
def get_project_skills_dirs() -> List[Path]:
|
||
"""Trusted project-local skill dirs for the current cwd (may be empty)."""
|
||
root = _current_project_root(trusted=True)
|
||
return _candidate_project_skills_dirs(root) if root is not None else []
|
||
|
||
|
||
def get_untrusted_project_skills_root() -> Optional[Tuple[Path, int]]:
|
||
"""(root, skill_count) when cwd's project has skills but is NOT trusted, else None."""
|
||
root = _current_project_root(trusted=False)
|
||
count = 0
|
||
for d in _candidate_project_skills_dirs(root) if root is not None else ():
|
||
try:
|
||
count += sum(1 for _ in iter_skill_index_files(d, "SKILL.md"))
|
||
except OSError:
|
||
continue
|
||
return (root, count) if count else None
|
||
|
||
|
||
# Scan-time injection defense: trust is a repo-level decision made once, but a
|
||
# `git pull` could inject a malicious skill into an already-trusted repo. Every
|
||
# project SKILL.md is scanned with the hub's skills_guard scanner (content-hash
|
||
# cached under HERMES_HOME, never inside the repo); "dangerous" excludes the
|
||
# skill from index, list, view and slash commands ("caution" loads, as on the hub).
|
||
|
||
# ── Project skill quarantine (scan-time injection defense) ──────────────── Trust (`hermes skills trust`)
|
||
# is a REPO-level decision made once; the repo's skill content keeps changing underneath it with every pull.
|
||
# The hub install path runs skills_guard on install, but project skills are read straight from a checkout —
|
||
# without this gate a `git pull` could inject a malicious skill into an already-trusted repo with no scan
|
||
# anywhere (#48974). Every project SKILL.md's parent dir is scanned with the same skills_guard scanner the
|
||
# hub uses (content-hash cached, so the cost is one scan per skill per content change). A "dangerous"
|
||
# verdict quarantines the skill: it is excluded from the index, skills_list, skill_view, and slash commands.
|
||
# "caution" loads (matches hub behavior for prose-level keyword hits) — the quarantine is for
|
||
# high-confidence findings only. The scan cache lives under HERMES_HOME, never inside the repo (we don't
|
||
# write artifacts into the user's checkout).
|
||
_PROJECT_SCAN_SOURCE = "project-local"
|
||
_PROJECT_QUARANTINE_CACHE: Dict[str, bool] = {} # skill_dir -> quarantined
|
||
|
||
|
||
def is_quarantined_project_skill(skill_md) -> bool:
|
||
"""True when a project skill's scan verdict is ``dangerous``. Fail-closed: a
|
||
scanner crash or missing scanner quarantines the skill. Scans
|
||
unconditionally — non-project callers should not call this."""
|
||
skill_dir = Path(skill_md).parent
|
||
try:
|
||
key = str(skill_dir.resolve())
|
||
except OSError:
|
||
key = str(skill_dir)
|
||
if key in _PROJECT_QUARANTINE_CACHE:
|
||
return _PROJECT_QUARANTINE_CACHE[key]
|
||
try:
|
||
from tools.skills_guard import scan_skill_cached
|
||
from hermes_constants import get_hermes_home
|
||
cache_dir = get_hermes_home() / "cache" / "project_skill_scans"
|
||
result, _prov = scan_skill_cached(skill_dir, source=_PROJECT_SCAN_SOURCE, cache_dir=cache_dir)
|
||
quarantined = result.verdict == "dangerous"
|
||
if quarantined:
|
||
logger.warning("Project skill quarantined (verdict=dangerous): %s — %s", skill_dir, result.summary)
|
||
except Exception:
|
||
logger.warning("Project skill scan failed — quarantining (fail closed): %s", skill_dir, exc_info=True)
|
||
quarantined = True
|
||
_PROJECT_QUARANTINE_CACHE[key] = quarantined
|
||
return quarantined
|
||
|
||
|
||
def iter_project_skill_files(project_dir: Path):
|
||
"""Yield non-quarantined SKILL.md files under a trusted project dir — the
|
||
single iteration chokepoint for the project tier, so the quarantine cannot
|
||
be bypassed by a call site forgetting the check."""
|
||
yield from (p for p in iter_skill_index_files(project_dir, "SKILL.md") if not is_quarantined_project_skill(p))
|
||
|
||
|
||
def normalize_skill_lookup_name(identifier: str) -> str:
|
||
"""Translate a trusted absolute skill path (slash commands / cron may store
|
||
them) to the relative form ``skill_view()`` accepts."""
|
||
raw_identifier = (identifier or "").strip()
|
||
if not raw_identifier:
|
||
return raw_identifier
|
||
identifier_path = Path(raw_identifier).expanduser()
|
||
if not identifier_path.is_absolute():
|
||
return raw_identifier.lstrip("/")
|
||
# Resolve the primary root via tools.skills_tool at CALL time: tests patch
|
||
# ``tools.skills_tool.SKILLS_DIR`` and skill_view() enforces ``_skills_dir()``
|
||
# (which follows the live profile-scoped HERMES_HOME), so normalization
|
||
# must agree with that exact root. Import deferred (cycle).
|
||
try:
|
||
# See #67277.
|
||
from tools import skills_tool as _skills_tool
|
||
primary_root = _skills_tool._skills_dir()
|
||
except Exception:
|
||
primary_root = get_skills_dir()
|
||
trusted_roots = [primary_root]
|
||
for getter in (get_project_skills_dirs, get_external_skills_dirs):
|
||
try:
|
||
trusted_roots.extend(getter())
|
||
except Exception:
|
||
pass
|
||
# Prefer the lexical path under a trusted root before resolving symlinks:
|
||
# ~/.hermes/skills/<name> may be a symlink to a checkout elsewhere, and
|
||
# resolving first would turn that trusted path into one skill_view rejects.
|
||
for root in trusted_roots:
|
||
if identifier_path.is_relative_to(root):
|
||
return str(identifier_path.relative_to(root))
|
||
try:
|
||
return str(identifier_path.resolve().relative_to(primary_root.resolve()))
|
||
except Exception:
|
||
logger.debug("Skill identifier %r is an absolute path outside trusted skills "
|
||
"roots — passing through unchanged (skill_view will reject it)", raw_identifier)
|
||
return raw_identifier
|
||
|
||
|
||
def _resolve_for_skill_ownership(path) -> Path:
|
||
path_obj = path if isinstance(path, Path) else Path(str(path))
|
||
try:
|
||
return path_obj.expanduser().resolve()
|
||
except (OSError, RuntimeError):
|
||
return path_obj.expanduser().absolute()
|
||
|
||
|
||
def is_external_skill_path(path) -> bool:
|
||
"""True when ``path`` lives under an external or trusted project skills dir.
|
||
Those are externally owned: autonomous lifecycle maintenance treats them as
|
||
read-only (user-directed tool calls may still edit them)."""
|
||
candidate = _resolve_for_skill_ownership(path)
|
||
roots: List[Path] = list(get_external_skills_dirs())
|
||
try:
|
||
roots.extend(get_project_skills_dirs())
|
||
except Exception:
|
||
pass
|
||
return any(candidate.is_relative_to(_resolve_for_skill_ownership(root)) for root in roots)
|
||
|
||
|
||
def _hermes_metadata(frontmatter: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""``metadata.hermes`` mapping from frontmatter, or ``{}`` when malformed."""
|
||
metadata = frontmatter.get("metadata")
|
||
hermes = metadata.get("hermes") if isinstance(metadata, dict) else None
|
||
return hermes if isinstance(hermes, dict) else {}
|
||
|
||
|
||
# ``session_platforms`` is the gateway-channel gate: session platforms the skill
|
||
# is FOR (hidden from the index elsewhere), unlike ``platforms:`` (host OS).
|
||
_CONDITION_KEYS = ("fallback_for_toolsets", "requires_toolsets", "fallback_for_tools", "requires_tools", "session_platforms")
|
||
|
||
|
||
def extract_skill_conditions(frontmatter: Dict[str, Any]) -> Dict[str, List]:
|
||
"""Extract conditional activation fields from parsed frontmatter (absent = ``[]``)."""
|
||
hermes = _hermes_metadata(frontmatter)
|
||
return {key: hermes.get(key, []) for key in _CONDITION_KEYS}
|
||
|
||
|
||
def extract_skill_config_vars(frontmatter: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||
"""Extract ``metadata.hermes.config`` declarations (key/description/default/prompt).
|
||
Entries missing ``key`` or ``description`` are skipped; ``prompt`` defaults to the description."""
|
||
raw = _hermes_metadata(frontmatter).get("config")
|
||
if isinstance(raw, dict):
|
||
raw = [raw]
|
||
if not raw or not isinstance(raw, list):
|
||
return []
|
||
result: Dict[str, Dict[str, Any]] = {}
|
||
for item in raw:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
key = str(item.get("key", "")).strip()
|
||
desc = str(item.get("description", "")).strip()
|
||
if not key or key in result or not desc:
|
||
continue
|
||
entry: Dict[str, Any] = {"key": key, "description": desc}
|
||
if item.get("default") is not None:
|
||
entry["default"] = item["default"]
|
||
prompt_text = item.get("prompt")
|
||
entry["prompt"] = prompt_text.strip() if isinstance(prompt_text, str) and prompt_text.strip() else desc
|
||
result[key] = entry
|
||
return list(result.values())
|
||
|
||
|
||
def discover_all_skill_config_vars() -> List[Dict[str, Any]]:
|
||
"""Config var declarations across all enabled, platform-compatible skills,
|
||
deduplicated by key; each dict carries a ``skill`` attribution key."""
|
||
all_vars: Dict[str, Dict[str, Any]] = {}
|
||
disabled = get_disabled_skill_names()
|
||
for skills_dir in get_all_skills_dirs():
|
||
if not skills_dir.is_dir():
|
||
continue
|
||
for skill_file in iter_skill_index_files(skills_dir, "SKILL.md"):
|
||
try:
|
||
frontmatter, _ = parse_frontmatter(skill_file.read_text(encoding="utf-8-sig"))
|
||
except Exception:
|
||
continue
|
||
skill_name = str(frontmatter.get("name") or skill_file.parent.name)
|
||
if skill_name in disabled or not skill_matches_platform(frontmatter):
|
||
continue
|
||
for var in extract_skill_config_vars(frontmatter):
|
||
if var["key"] not in all_vars:
|
||
var["skill"] = skill_name
|
||
all_vars[var["key"]] = var
|
||
return list(all_vars.values())
|
||
|
||
|
||
# Skill config vars are stored under skills.config.<logical key> in config.yaml.
|
||
SKILL_CONFIG_PREFIX = "skills.config"
|
||
|
||
|
||
def _resolve_dotpath(config: Dict[str, Any], dotted_key: str):
|
||
"""Walk a nested dict following a dotted key; None if any part is missing."""
|
||
current = config
|
||
for part in dotted_key.split("."):
|
||
if not isinstance(current, dict) or part not in current:
|
||
return None
|
||
current = current[part]
|
||
return current
|
||
|
||
|
||
_HOME_VAR_RE = re.compile(r"\$(?:\{HOME\}|HOME)(?=$|[/\\])")
|
||
|
||
|
||
def _expand_skill_config_path(value: str) -> str:
|
||
"""Expand ``~`` / ``$HOME`` against the HOME Hermes injects into tool subprocesses.
|
||
|
||
Skill config defaults describe paths the agent hands to tools, so in a container where the
|
||
control process HOME (``/opt/data``) differs from the tool HOME (``{HERMES_HOME}/home``) a
|
||
plain ``expanduser`` pointed the prompt at a path no tool would ever read (#12260).
|
||
"""
|
||
subprocess_home = get_subprocess_home()
|
||
if subprocess_home:
|
||
if value == "~" or value.startswith(("~/", "~\\")):
|
||
value = subprocess_home + value[1:]
|
||
# Callable replacement: a literal template would parse backslashes in the home path
|
||
# as regex escapes.
|
||
value = _HOME_VAR_RE.sub(lambda _m: subprocess_home, value)
|
||
return os.path.expanduser(os.path.expandvars(value))
|
||
|
||
|
||
def resolve_skill_config_values(config_vars: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||
"""Map logical skill config keys to current values (or declared defaults);
|
||
path-like string values are ``~``/``$HOME``/``${VAR}`` expanded against the tool HOME."""
|
||
config = _load_raw_config()
|
||
resolved: Dict[str, Any] = {}
|
||
for var in config_vars:
|
||
value = _resolve_dotpath(config, f"{SKILL_CONFIG_PREFIX}.{var['key']}")
|
||
if value is None or (isinstance(value, str) and not value.strip()):
|
||
value = var.get("default", "")
|
||
if isinstance(value, str) and ("~" in value or "$" in value):
|
||
value = _expand_skill_config_path(value)
|
||
resolved[var["key"]] = value
|
||
return resolved
|
||
|
||
|
||
SKILL_PROMPT_DESC_LIMIT = 60
|
||
|
||
|
||
def _normalize_skill_description(frontmatter: Dict[str, Any]) -> str:
|
||
"""Normalize a skill's description field for comparison/truncation."""
|
||
raw_desc = frontmatter.get("description", "")
|
||
return str(raw_desc).strip().strip("'\"") if raw_desc else ""
|
||
|
||
|
||
def extract_skill_description(frontmatter: Dict[str, Any]) -> str:
|
||
"""Extract a system-prompt-length description from parsed frontmatter."""
|
||
desc = _normalize_skill_description(frontmatter)
|
||
return desc[:SKILL_PROMPT_DESC_LIMIT - 3] + "..." if len(desc) > SKILL_PROMPT_DESC_LIMIT else desc
|
||
|
||
|
||
def is_skill_description_truncated_for_prompt(frontmatter: Dict[str, Any]) -> bool:
|
||
"""True when the description will be truncated in the system prompt skill index."""
|
||
return len(_normalize_skill_description(frontmatter)) > SKILL_PROMPT_DESC_LIMIT
|
||
|
||
|
||
def iter_skill_index_files(skills_dir: Path, filename: str):
|
||
"""Walk skills_dir yielding sorted paths matching *filename*; prunes
|
||
EXCLUDED_SKILL_DIRS and support dirs of skill roots. Org mirrors are
|
||
TOKEN-GATED: only the active org's subdir is walked, so leaving an org
|
||
stops its skills resolving without manual cleanup."""
|
||
skills_dir_str = str(skills_dir)
|
||
active_org = read_active_org_id(skills_dir)
|
||
org_root = os.path.join(skills_dir_str, ORG_MIRROR_DIR_NAME)
|
||
matches: list[str] = []
|
||
for root, dirs, files in os.walk(skills_dir_str, followlinks=True):
|
||
has_skill_md = "SKILL.md" in files
|
||
if root == skills_dir_str and ORG_MIRROR_DIR_NAME in dirs and active_org is None:
|
||
dirs.remove(ORG_MIRROR_DIR_NAME)
|
||
elif root == org_root:
|
||
dirs[:] = [d for d in dirs if d == active_org]
|
||
dirs[:] = [d for d in dirs if d not in EXCLUDED_SKILL_DIRS and not (has_skill_md and d in SKILL_SUPPORT_DIRS)]
|
||
if filename in files:
|
||
matches.append(os.path.join(root, filename))
|
||
yield from map(Path, sorted(matches))
|
||
|
||
|
||
# Namespace helpers for plugin-provided skills.
|
||
_NAMESPACE_RE = re.compile(r"^[a-zA-Z0-9_-]+$")
|
||
|
||
|
||
def parse_qualified_name(name: str) -> Tuple[Optional[str], str]:
|
||
"""Split ``'namespace:skill-name'`` into ``(namespace, bare_name)``; ``(None, name)`` without ``':'``."""
|
||
namespace, sep, bare = name.partition(":")
|
||
return (namespace, bare) if sep else (None, name)
|
||
|
||
|
||
def is_valid_namespace(candidate: Optional[str]) -> bool:
|
||
"""Check whether *candidate* is a valid namespace (``[a-zA-Z0-9_-]+``)."""
|
||
return bool(candidate) and bool(_NAMESPACE_RE.match(candidate))
|
||
|
||
|
||
# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
|
||
# Names external plugins imported from this module before the Sep 2026 decomposition.
|
||
# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
|
||
# The whole block is removed by reverting the commit that added it.
|
||
|
||
def get_scan_ordered_skills_dirs() -> List[Path]:
|
||
"""All skill dirs in precedence order: project → local → external.
|
||
|
||
First-wins name deduplication over this order gives project skills
|
||
priority over profile-local and external ones.
|
||
"""
|
||
dirs = list(get_project_skills_dirs())
|
||
dirs.extend(get_all_skills_dirs())
|
||
return dirs
|
||
# ---- END PLUGIN-COMPAT ----
|