From d2f2862d514de79d6f0342e2c22d7ff87235e849 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Thu, 3 Sep 2026 01:09:55 -0700 Subject: [PATCH] refactor(tools): fold single-use readiness helpers, compact module docstrings, wrap long lines --- tools/skills_sync.py | 48 +++++++++-------- tools/skills_sync_bundled_ops.py | 4 +- tools/skills_sync_optional.py | 24 +++++---- tools/skills_tool.py | 89 ++++++++++++++++---------------- tools/skills_tool_dedup.py | 9 ++-- tools/skills_tool_plugin.py | 26 +++++----- tools/skills_tool_setup.py | 52 +++++++------------ 7 files changed, 118 insertions(+), 134 deletions(-) diff --git a/tools/skills_sync.py b/tools/skills_sync.py index 275ea16668..be429cbd87 100644 --- a/tools/skills_sync.py +++ b/tools/skills_sync.py @@ -1,12 +1,9 @@ #!/usr/bin/env python3 -"""Skills Sync -- manifest-based seeding and updating of bundled skills. - -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. -""" +"""Skills Sync -- manifest-based seeding and updating of bundled skills. 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 import logging @@ -61,10 +58,6 @@ def _manifest_file() -> Path: return _live(MANIFEST_FILE, _MANIFEST_FILE_AT_IMPORT, lambda: _skills_dir() / ".bundled_manifest") -def _rel_skills_posix(path: Path) -> str: - return path.relative_to(_skills_dir()).as_posix() - - # 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" @@ -78,6 +71,10 @@ def _get_optional_dir() -> Path: return get_optional_skills_dir(Path(__file__).parent.parent / "optional-skills") +def _rel_skills_posix(path: Path) -> str: + return path.relative_to(_skills_dir()).as_posix() + + def _iter_skill_mds(root: Path, sort: bool = False) -> Iterator[Path]: """Yield every non-excluded SKILL.md under ``root`` (nothing when it does not exist).""" found = root.rglob("SKILL.md") if root.exists() else iter(()) @@ -123,15 +120,16 @@ def _write_manifest(entries: Dict[str, str]): """Atomic v2 write, preserving an existing file's mode/owner (not mkstemp's 0600).""" _manifest_file().parent.mkdir(parents=True, exist_ok=True) try: - atomic_write_text(_manifest_file(), "".join(f"{n}:{h}\n" for n, h in sorted(entries.items())), - tmp_prefix=".bundled_manifest_", preserve_mode=True) + data = "".join(f"{n}:{h}\n" for n, h in sorted(entries.items())) + atomic_write_text(_manifest_file(), data, tmp_prefix=".bundled_manifest_", preserve_mode=True) except Exception as e: logger.debug("Failed to write skills manifest %s: %s", _manifest_file(), e, exc_info=True) def _discover_bundled_skills(bundled_dir: Path) -> List[Tuple[str, Path]]: """``(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``.""" + 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 [ @@ -169,8 +167,7 @@ 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 (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 stay. Returns the rel source path on move.""" + byte-identical to the origin hash — proof *we* placed it — moves. Returns rel source path.""" origin_hash = st.manifest.get(skill_name, "") if not origin_hash: return None @@ -261,8 +258,9 @@ def _install_new_skill(st: _SyncState, skill_name: str, skill_src: Path, dest: P st.manifest[skill_name] = bundled_hash else: st.say( - f" ⚠ {skill_name}: bundled version shipped but you already have a local skill by this name — " - f"yours was kept. Run `hermes skills reset {skill_name}` to replace it with the bundled version.") + f" ⚠ {skill_name}: bundled version shipped but you already have a local skill " + f"by this name — yours was kept. Run `hermes skills reset {skill_name}` to " + f"replace it with the bundled version.") else: _copy_dir(skill_src, dest) st.copied.append(skill_name) @@ -273,7 +271,7 @@ def _install_new_skill(st: _SyncState, skill_name: str, skill_src: Path, dest: P def _replace_skill_dir(skill_src: Path, dest: Path) -> None: - """Replace ``dest`` with a fresh copy of ``skill_src`` via a ``.bak`` sibling, restoring on failure.""" + """Replace ``dest`` with a fresh copy of ``skill_src`` via a .bak sibling; restore on failure.""" backup = dest.with_suffix(".bak") if backup.exists(): # a stale .bak would make shutil.move() nest dest INSIDE it _rmtree_writable(backup) @@ -286,7 +284,8 @@ def _replace_skill_dir(skill_src: Path, dest: Path) -> None: try: _rmtree_writable(dest) except OSError: - logger.warning("Could not clear partial copy %s during restore", dest, exc_info=True) + logger.warning("Could not clear partial copy %s during restore", dest, + exc_info=True) if not dest.exists(): shutil.move(str(backup), str(dest)) raise @@ -303,7 +302,7 @@ def _update_existing_skill(st: _SyncState, skill_name: str, skill_src: Path, des st.skipped += 1 return user_hash = _dir_hash(dest) - if not origin_hash: # v1 migration: baseline from the user's copy (can't tell edit from upstream change) + if not origin_hash: # v1 migration: baseline from user's copy (can't tell edit from upstream) st.manifest[skill_name] = user_hash st.skipped += 1 return @@ -389,14 +388,13 @@ def sync_skills(quiet: bool = False) -> dict: "total_bundled": len(bundled_skills), "optional_provenance_backfilled": _backfill_optional_provenance(quiet=quiet), "shadowed_by_external": st.shadowed_by_external, - "skipped_opt_out": essential_only} # lets callers report "opted out" rather than a normal sync + "skipped_opt_out": essential_only} # lets callers report "opted out", not a normal sync def _rmtree_writable(path: Path) -> None: """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``.""" + child of the active skills root (bad join / missing HERMES_HOME / malicious manifest entry).""" target = Path(path).resolve() skills_root = _skills_dir().resolve() if skills_root not in target.parents: diff --git a/tools/skills_sync_bundled_ops.py b/tools/skills_sync_bundled_ops.py index 930f707701..f3d0e922be 100644 --- a/tools/skills_sync_bundled_ops.py +++ b/tools/skills_sync_bundled_ops.py @@ -9,7 +9,7 @@ from tools.skills_sync_optional import _skill_file_list, _ss def _is_tracked_user_modification(origin_hash: str, user_hash: str) -> bool: """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-modified so they never drift.""" + don't count) AND differing content. Shared by the sync loop and list-modified (no drift).""" return bool(origin_hash) and user_hash != origin_hash @@ -140,7 +140,7 @@ _OPT_OUT_MESSAGES = { # (enabled, changed) -> message def set_bundled_skills_opt_out(enabled: bool) -> dict: """Toggle the .no-bundled-skills marker (on-disk half of ``hermes skills opt-out`` / ``opt-in``; - removing present skills is ``remove_pristine_bundled_skills``). ``{ok, changed, marker, message}``.""" + removing present skills is ``remove_pristine_bundled_skills``).""" ss = _ss() marker = ss._hermes_home() / ss.NO_BUNDLED_SKILLS_MARKER existed = marker.exists() diff --git a/tools/skills_sync_optional.py b/tools/skills_sync_optional.py index 2dd60ca5dc..01c713ee99 100644 --- a/tools/skills_sync_optional.py +++ b/tools/skills_sync_optional.py @@ -1,5 +1,5 @@ -"""Official optional-skill provenance: hub-lock backfill and restore. Profile-scoped paths -and patchable helpers resolve through ``_ss()`` at call time (``tools.skills_sync`` patches work).""" +"""Official optional-skill provenance: hub-lock backfill and restore. Profile-scoped paths and +patchable helpers resolve through ``_ss()`` at call time so ``tools.skills_sync`` patches work.""" import json import logging @@ -21,7 +21,7 @@ def _ss(): def _content_hash(directory: Path) -> str: - """Hub-lock hash style; provenance metadata only, so fall back to local MD5 without guard deps.""" + """Hub-lock hash style; provenance metadata only, so fall back to local MD5 sans guard deps.""" try: from tools.skills_guard import content_hash return content_hash(directory) @@ -74,14 +74,15 @@ 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 (hub-lock - slug or 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]] = {} if optional_dir.exists(): for skill_md, src, install_path in _iter_optional_skills(optional_dir, root_relative=True): - index[src.name] = index[ss._read_skill_name(skill_md, src.name)] = (src.name, install_path, src) + value = (src.name, install_path, src) + index[src.name] = index[ss._read_skill_name(skill_md, src.name)] = value return index @@ -192,15 +193,16 @@ def _backfill_optional_provenance(quiet: bool = False) -> List[str]: continue timestamp = datetime.now(timezone.utc).isoformat() installed[lock_name] = { - "source": "official", "identifier": f"official/{install_path}", "trust_level": "builtin", - "scan_verdict": "backfilled", "content_hash": _content_hash(dest), "install_path": install_path, + "source": "official", "identifier": f"official/{install_path}", + "trust_level": "builtin", "scan_verdict": "backfilled", + "content_hash": _content_hash(dest), "install_path": install_path, "files": _skill_file_list(dest), "metadata": {"backfilled_from": "optional-skills"}, "installed_at": timestamp, "updated_at": timestamp} existing_paths.add(install_path) backfilled.append(lock_name) if not quiet: print(f" = {lock_name} (official optional provenance backfilled)") - if backfilled: # atomic: a crash mid-write must not wipe all provenance (reader resets on bad JSON) - atomic_write_text(ss._skills_dir() / ".hub" / "lock.json", - json.dumps(data, indent=2, ensure_ascii=False) + "\n", tmp_prefix=".lock_") + if backfilled: # atomic: a mid-write crash must not wipe provenance (reader resets bad JSON) + atomic_write_text(ss._skills_dir() / ".hub" / "lock.json", tmp_prefix=".lock_", + content=json.dumps(data, indent=2, ensure_ascii=False) + "\n") return backfilled diff --git a/tools/skills_tool.py b/tools/skills_tool.py index 26a910da99..431b02bf79 100644 --- a/tools/skills_tool.py +++ b/tools/skills_tool.py @@ -1,11 +1,8 @@ #!/usr/bin/env python3 -"""Skills Tool — list and view skill documents (progressive disclosure). - -A skill is a directory holding SKILL.md (YAML frontmatter + instructions) plus optional -references/, templates/, assets/, scripts/. `skills_list` returns name/description only; -`skill_view` returns full content and linked files. Sibling modules (skills_tool_setup / -_plugin / _dedup) have every name re-exported here so imports and patches keep working. -""" +"""Skills Tool — list and view skill documents (progressive disclosure). A skill is a directory +holding SKILL.md (YAML frontmatter + instructions) plus optional references/, templates/, assets/, +scripts/. `skills_list` returns name/description only; `skill_view` returns full content and +linked files. Sibling modules (skills_tool_setup / _plugin / _dedup) re-export here.""" import json import logging @@ -22,13 +19,13 @@ from agent.skill_utils import ( 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, _build_setup_note, _capture_required_environment_variables, - _get_required_environment_variables, _get_terminal_backend_name, _is_env_var_persisted, - _is_remote_env_backend) + _get_required_environment_variables, _is_env_var_persisted, _is_remote_env_backend) from tools.skills_tool_plugin import ( # noqa: F401 MAX_DESCRIPTION_LENGTH, MAX_NAME_LENGTH, _INJECTION_PATTERNS, _fail, _json, _mark_background_review_read, _preprocess_skill, _read_skill_text, _safe_frontmatter, _serve_plugin_skill, _serve_skill_file, _truncate_description) -from tools.skills_tool_dedup import _check_skill_view_dedup, _record_skill_view, reset_skill_view_dedup # noqa: F401 +from tools.skills_tool_dedup import ( # noqa: F401 + _check_skill_view_dedup, _record_skill_view, reset_skill_view_dedup) logger = logging.getLogger(__name__) @@ -44,6 +41,7 @@ def _skills_scan_signature(dirs_to_scan, disabled) -> tuple: """O(#dirs + #categories) stat-based change signature; platform is read via ``agent.skill_utils.sys`` so test patches are honored.""" from agent import skill_utils as _skill_utils + platform = getattr(getattr(_skill_utils, "sys", None), "platform", "") sig = [] for d in dirs_to_scan: try: @@ -56,10 +54,10 @@ def _skills_scan_signature(dirs_to_scan, disabled) -> tuple: 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), getattr(getattr(_skill_utils, "sys", None), "platform", "")) + return (tuple(sig), frozenset(disabled), platform) -HERMES_HOME = get_hermes_home() # all skills live in ~/.hermes/skills/ (seeded from bundled skills/) +HERMES_HOME = get_hermes_home() # all skills live in ~/.hermes/skills/ (seeded from bundled) SKILLS_DIR = HERMES_HOME / "skills" _SKILLS_DIR_AT_IMPORT = SKILLS_DIR @@ -154,15 +152,6 @@ def _parse_tags(tags_value) -> List[str]: return [t.strip().strip("\"'") for t in tags_value.split(",") if t.strip()] -def _get_session_platform() -> str: - """Platform from gateway session context (mirrors ``get_disabled_skill_names``).""" - try: - from gateway.session_context import get_session_env - return get_session_env("HERMES_SESSION_PLATFORM") or "" - except Exception: - return "" - - def _is_skill_disabled(name: str, platform: str = None) -> bool: """Disabled in config? Platform precedence: explicit arg, ``HERMES_PLATFORM``, session ``HERMES_SESSION_PLATFORM``. A globally-disabled skill stays disabled on every platform @@ -170,9 +159,14 @@ def _is_skill_disabled(name: str, platform: str = None) -> bool: try: from hermes_cli.config import load_config skills_cfg = load_config().get("skills", {}) - resolved_platform = platform or os.getenv("HERMES_PLATFORM") or _get_session_platform() - platform_disabled = ( - cfg_get(skills_cfg, "platform_disabled", resolved_platform) if resolved_platform else None) + resolved_platform = platform or os.getenv("HERMES_PLATFORM") + if not resolved_platform: + with suppress(Exception): + from gateway.session_context import get_session_env + resolved_platform = get_session_env("HERMES_SESSION_PLATFORM") or "" + platform_disabled = None + if resolved_platform: + platform_disabled = cfg_get(skills_cfg, "platform_disabled", resolved_platform) in_platform = platform_disabled is not None and name in platform_disabled return in_platform or name in skills_cfg.get("disabled", []) except Exception: @@ -263,7 +257,8 @@ def skills_list(category: str = None, task_id: str = None) -> str: all_skills = _sort_skills(all_skills) categories = sorted({s.get("category") for s in all_skills if s.get("category")}) return _json({ - "success": True, "skills": all_skills, "categories": categories, "count": len(all_skills), + "success": True, "skills": all_skills, "categories": categories, + "count": len(all_skills), "hint": "Use skill_view(name) to see full content, tags, and linked files"}) except Exception as e: return tool_error(str(e), success=False) @@ -271,9 +266,8 @@ def skills_list(category: str = None, task_id: str = None) -> str: def _resolve_plugin_skill(name, file_path, task_id, preprocess): """``plugin:skill`` dispatch: ``(result_json, None)`` when answered, else ``(None, - local_category_name)`` to fall through to the flat-tree scan — categorized local skills - also use ``category:skill`` in config/gateway prompts, so the on-disk ``category/skill`` - form is returned (None without a bare part).""" + local_category_name)`` to fall through to the flat-tree scan — categorized local skills also use + ``category:skill`` in config/gateway prompts, so the on-disk ``category/skill`` form returns.""" from agent.skill_utils import is_valid_namespace, parse_qualified_name from hermes_cli.plugins import discover_plugins, get_plugin_manager namespace, bare = parse_qualified_name(name) @@ -352,7 +346,8 @@ def _collect_skill_candidates(name, local_category_name, all_dirs): # 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 or _safe_frontmatter(found_skill_md).get("name") == name: + if (found_skill_md.parent.name == name + or _safe_frontmatter(found_skill_md).get("name") == name): _record(found_skill_md.parent, found_skill_md) # Legacy flat .md anywhere under the dir; support docs are excluded # (they load via file_path and must not shadow real skills sharing the basename). @@ -377,7 +372,8 @@ def _skill_linked_files(skill_dir: Optional[Path]) -> dict: base = skill_dir / sub found = [ str(f.relative_to(skill_dir)) for g in globs if base.exists() - for f in (base.rglob(g) if recursive else base.glob(g)) if not files_only or f.is_file()] + 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 @@ -393,7 +389,8 @@ def _org_provenance_header(skill_dir: Path, active_skills_dir: Path): prov: dict = {} if prov_org: with suppress(Exception): - loaded = json.loads(_read_skill_text(active_skills_dir / "_org" / prov_org / ORG_PROVENANCE_FILE)) + prov_path = active_skills_dir / "_org" / prov_org / ORG_PROVENANCE_FILE + loaded = json.loads(_read_skill_text(prov_path)) prov = loaded if isinstance(loaded, dict) else {} author = str(prov.get("author_device") or prov.get("author_user_id") or "") ts = str(prov.get("ts") or "") @@ -415,7 +412,7 @@ def _skill_readiness(frontmatter: Dict[str, Any], skill_name: str) -> Tuple[dict allows) and register what's available for sandboxes. Returns ``(fields, extras)``: fields go before ``_source_path`` in the skill_view result, extras after — key order is tool output.""" required_env_vars = _get_required_environment_variables(frontmatter) - backend = _get_terminal_backend_name() + backend = str(os.getenv("TERMINAL_ENV", "local")).strip().lower() or "local" env_snapshot = load_env() missing_required_env_vars = [ e for e in required_env_vars @@ -449,9 +446,10 @@ def _skill_readiness(frontmatter: Dict[str, Any], skill_name: str) -> Tuple[dict status = SkillReadinessStatus.SETUP_NEEDED if setup_needed else SkillReadinessStatus.AVAILABLE fields = { "required_environment_variables": required_env_vars, "required_commands": [], - "missing_required_environment_variables": remaining, "missing_credential_files": missing_cred_files, - "missing_required_commands": [], "setup_needed": setup_needed, - "setup_skipped": capture_result["setup_skipped"], "readiness_status": status.value} + "missing_required_environment_variables": remaining, + "missing_credential_files": missing_cred_files, "missing_required_commands": [], + "setup_needed": setup_needed, "setup_skipped": capture_result["setup_skipped"], + "readiness_status": status.value} extras: dict = {} if setup_help := next((e["help"] for e in required_env_vars if e.get("help")), None): extras["setup_help"] = setup_help @@ -465,11 +463,12 @@ def _skill_readiness(frontmatter: Dict[str, Any], skill_name: str) -> Tuple[dict return fields, extras -def _locate_skill(name: str, local_category_name: Optional[str], project_dirs: list, all_dirs: list): +def _locate_skill(name: str, local_category_name: Optional[str], project_dirs: list, all_dirs): """Unique on-disk skill for *name*: collision refusal, project-tier precedence, quarantine - gate, not-found listing. ``(error_json, skill_dir, skill_md)``; skill_md set when error is None.""" + gate, not-found listing. ``(error_json, skill_dir, skill_md)``; skill_md set iff no error.""" if not all_dirs: - return _fail("Skills directory does not exist yet. It will be created on first install."), None, None + return _fail( + "Skills directory does not exist yet. It will be created on first install."), None, None 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; @@ -493,8 +492,8 @@ def _locate_skill(name: str, local_category_name: Optional[str], project_dirs: l 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`." - ), None, None + hint="Inspect the skill in the repo checkout, or untrust the repo with " + "`hermes skills untrust`."), None, None 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, @@ -502,7 +501,7 @@ def _locate_skill(name: str, local_category_name: Optional[str], project_dirs: l return None, skill_dir, skill_md -def _log_security_warnings(name: str, skill_md: Path, content: str, all_dirs: list, active_skills_dir: Path): +def _log_security_warnings(name: str, skill_md: Path, content: str, all_dirs, active_skills_dir): """Warn (never block) when loaded from outside the trusted dirs (project + local + external) and/or when common prompt-injection patterns appear.""" trusted_dirs = [active_skills_dir.resolve()] @@ -538,7 +537,8 @@ def skill_view( if local_category_name and (lookup_error := _skill_lookup_path_error(local_category_name)): return _fail(lookup_error, hint=_LOOKUP_HINT) project_dirs, all_dirs, active_skills_dir = _skill_search_dirs() - error, skill_dir, skill_md = _locate_skill(name, local_category_name, project_dirs, all_dirs) + error, skill_dir, skill_md = _locate_skill( + name, local_category_name, project_dirs, all_dirs) if error is not None: return error try: # read once — reused for platform check and main content @@ -578,8 +578,9 @@ def skill_view( logger.debug("Could not resolve org provenance for %s", skill_name, exc_info=True) result = { "success": True, "name": skill_name, "description": frontmatter.get("description", ""), - "tags": tags, "related_skills": related_skills, "content": header + rendered_content, "path": rel_path, - "skill_dir": str(skill_dir) if skill_dir else None, "org_provenance": org_provenance, + "tags": tags, "related_skills": related_skills, "content": header + rendered_content, + "path": rel_path, "skill_dir": str(skill_dir) if skill_dir else None, + "org_provenance": org_provenance, "linked_files": linked_files if linked_files else None, "usage_hint": "To view linked files, call skill_view(name, file_path) where file_path is e.g. 'references/api.md' or 'assets/config.yaml'" if linked_files else None, **readiness, diff --git a/tools/skills_tool_dedup.py b/tools/skills_tool_dedup.py index 7796f0ed77..ef78292741 100644 --- a/tools/skills_tool_dedup.py +++ b/tools/skills_tool_dedup.py @@ -35,7 +35,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.""" # 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 not task_id or payload.get("setup_needed") or payload.get("readiness_status") == "setup_needed": + if (not task_id or payload.get("setup_needed") + or payload.get("readiness_status") == "setup_needed"): return if (fp := _skill_view_fingerprint(payload)) is None: return @@ -73,9 +74,9 @@ def _check_skill_view_dedup(task_id, name, file_path) -> str | None: cache.pop(key, None) return None return json.dumps({ - "success": True, "status": "unchanged", "name": rec_name, "file": file_path or "SKILL.md", - "dedup": True, "content_returned": False, "message": _SKILL_VIEW_DEDUP_MESSAGE}, - ensure_ascii=False) + "success": True, "status": "unchanged", "name": rec_name, + "file": file_path or "SKILL.md", "dedup": True, "content_returned": False, + "message": _SKILL_VIEW_DEDUP_MESSAGE}, ensure_ascii=False) return None diff --git a/tools/skills_tool_plugin.py b/tools/skills_tool_plugin.py index f0b2a7bbd2..a8b31b64bb 100644 --- a/tools/skills_tool_plugin.py +++ b/tools/skills_tool_plugin.py @@ -1,9 +1,6 @@ -"""Plugin-provided skill serving for ``skill_view`` (``plugin:skill`` names) plus the -JSON / file-serving helpers shared with the local-skill path. - -Helpers tests patch on the origin module (``_is_skill_disabled``, ``_parse_frontmatter``, -``skill_matches_platform``) are looked up lazily via ``tools.skills_tool``. -""" +"""Plugin-provided skill serving for ``skill_view`` (``plugin:skill`` names) plus the JSON / +file-serving helpers shared with the local-skill path. Helpers tests patch on the origin module +(``_is_skill_disabled``, ``_parse_frontmatter``, ``skill_matches_platform``) resolve lazily.""" import json import logging @@ -34,8 +31,8 @@ def _fail(error: str, **extra) -> str: def _read_skill_text(path: Path) -> str: - """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.""" + """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") @@ -84,8 +81,9 @@ def _serve_skill_file( return _fail(path_error, **extra) # is_file(), not exists(): a bare directory must take the not-found branch. if not target.is_file(): - listing = {"available_files": _available_skill_files(skill_root), - "hint": "Use one of the available file paths listed above"} if list_available else {} + listing = {} if not list_available else { + "available_files": _available_skill_files(skill_root), + "hint": "Use one of the available file paths listed above"} return _fail(f"File '{file_path}' not found in skill '{label}'.", **listing) try: content = _read_skill_text(target) @@ -150,11 +148,11 @@ def _serve_plugin_skill( with suppress(Exception): # 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." + ( - f"\nSibling skills: {', '.join(siblings)}.\n" - f"Use qualified form to invoke siblings (e.g. {namespace}:{siblings[0]})." if siblings else "" - ) + "]\n\n" + f"\nSibling skills: {', '.join(siblings)}.\nUse qualified form to invoke siblings " + f"(e.g. {namespace}:{siblings[0]})." if siblings else "") + "]\n\n" rendered_content = content if not preprocess else _preprocess_skill( - content, skill_md.parent, session_id, "Could not preprocess plugin skill %s:%s", namespace, bare) + content, skill_md.parent, session_id, "Could not preprocess plugin skill %s:%s", + namespace, bare) return _json({ "success": True, "name": qualified_name, "content": banner + rendered_content, "description": _truncate_description(str(parsed_frontmatter.get("description", ""))), diff --git a/tools/skills_tool_setup.py b/tools/skills_tool_setup.py index 2b6e6f0bff..243f7b0eb6 100644 --- a/tools/skills_tool_setup.py +++ b/tools/skills_tool_setup.py @@ -1,9 +1,6 @@ -"""Skill readiness: required env vars, secret capture, and setup notes. - -Split out of ``tools.skills_tool`` (names 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. -""" +"""Skill readiness: required env vars, secret capture, and setup notes. Split out of +``tools.skills_tool`` (names re-imported there); module state (``_secret_capture_callback``, +``load_env``) stays there and is read lazily at call time so origin-module patches are honored.""" import logging import os @@ -37,20 +34,9 @@ def _is_remote_env_backend(backend: str) -> bool: return False -def _legacy_env_vars(frontmatter: Dict[str, Any]) -> List[str]: - """``prerequisites.env_vars`` (legacy form): a single string or a list.""" - prereqs = frontmatter.get("prerequisites") - value = prereqs.get("env_vars") if isinstance(prereqs, dict) else None - if not value: - return [] - return [str(item) for item in ([value] if isinstance(value, str) else value) if str(item).strip()] - - def _as_dict_list(raw: Any) -> list: """Accept a single mapping or a list; anything else is treated as empty.""" - if isinstance(raw, dict): - return [raw] - return raw if isinstance(raw, list) else [] + return [raw] if isinstance(raw, dict) else raw if isinstance(raw, list) else [] def _clean_str(value: Any) -> str | None: @@ -64,15 +50,19 @@ def _get_required_environment_variables(frontmatter: Dict[str, Any]) -> List[Dic setup = frontmatter.get("setup") setup = setup if isinstance(setup, dict) else {} setup_help = _clean_str(setup.get("help")) + prereqs = frontmatter.get("prerequisites") + legacy = (prereqs.get("env_vars") if isinstance(prereqs, dict) else None) or [] + legacy = [legacy] if isinstance(legacy, str) else legacy required: Dict[str, Dict[str, Any]] = {} # env name -> entry, insertion-ordered, first wins declared = _as_dict_list(frontmatter.get("required_environment_variables")) - entries = [{"name": i} if isinstance(i, str) else i for i in declared if isinstance(i, (str, dict))] + entries = [{"name": i} if isinstance(i, str) else i for i in declared + if isinstance(i, (str, dict))] # collect_secrets entries: env_var is the name; provider_url (or url) doubles as help. entries += [ {"name": i.get("env_var"), "prompt": i.get("prompt"), "url": str(i.get("provider_url") or i.get("url") or "").strip() or None} for i in _as_dict_list(setup.get("collect_secrets")) if isinstance(i, dict)] - entries += [{"name": v} for v in _legacy_env_vars(frontmatter)] + entries += [{"name": str(v)} for v in legacy if str(v).strip()] for entry in entries: env_name = str(entry.get("name") or entry.get("env_var") or "").strip() if not env_name or env_name in required or not _ENV_VAR_NAME_RE.match(env_name): @@ -106,7 +96,12 @@ def _capture_required_environment_variables( # 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. if _is_gateway_surface() and not env_var_enabled("HERMES_INTERACTIVE"): - return _capture_result(missing_names, gateway_setup_hint=_gateway_setup_hint()) + try: + from gateway.platforms.base import GATEWAY_SECRET_CAPTURE_UNSUPPORTED_MESSAGE as hint + except Exception: + hint = (f"Secure secret entry is not available. Load this skill in the local CLI to be " + f"prompted, or add the key to {display_hermes_home()}/.env manually.") + return _capture_result(missing_names, gateway_setup_hint=hint) if (callback := _st._secret_capture_callback) is None: return _capture_result(missing_names) remaining_names: List[str] = [] @@ -130,25 +125,14 @@ def _is_gateway_surface() -> bool: return bool(get_session_env("HERMES_SESSION_PLATFORM")) -def _get_terminal_backend_name() -> str: - return str(os.getenv("TERMINAL_ENV", "local")).strip().lower() or "local" - - def _is_env_var_persisted(var_name: str, env_snapshot: Dict[str, str]) -> bool: """Set (non-empty) in the .env snapshot, else in the process environment.""" return bool(env_snapshot[var_name] if var_name in env_snapshot else os.getenv(var_name)) -def _gateway_setup_hint() -> str: - try: - from gateway.platforms.base import GATEWAY_SECRET_CAPTURE_UNSUPPORTED_MESSAGE - return GATEWAY_SECRET_CAPTURE_UNSUPPORTED_MESSAGE - except Exception: - return f"Secure secret entry is not available. Load this skill in the local CLI to be prompted, or add the key to {display_hermes_home()}/.env manually." - - def _build_setup_note( - readiness_status: SkillReadinessStatus, missing: List[str], setup_help: str | None = None) -> str | None: + readiness_status: SkillReadinessStatus, missing: List[str], + setup_help: str | None = None) -> str | None: if readiness_status != SkillReadinessStatus.SETUP_NEEDED: return None note = f"Setup needed before using this skill: missing {', '.join(missing) if missing else 'required prerequisites'}."