diff --git a/tests/tools/test_skill_linter.py b/tests/tools/test_skill_linter.py index b4c03692b2..538bcd9a2a 100644 --- a/tests/tools/test_skill_linter.py +++ b/tests/tools/test_skill_linter.py @@ -7,8 +7,6 @@ import pytest from tools.skill_linter import ( ERROR, WARNING, - format_findings, - has_errors, lint_content, lint_skill, ) @@ -105,7 +103,7 @@ def test_bad_name_format_is_error(): content = CLEAN.replace("name: my-skill", "name: My_Skill!") findings = lint_content(content) assert "name-format" in _rules(findings) - assert has_errors(findings) + assert any(f.severity == ERROR for f in findings) def test_name_dir_mismatch_is_error(tmp_path): @@ -113,7 +111,7 @@ def test_name_dir_mismatch_is_error(tmp_path): skill_dir.mkdir() findings = lint_content(CLEAN, skill_dir=skill_dir) # name is my-skill assert "name-dir-mismatch" in _rules(findings) - assert has_errors(findings) + assert any(f.severity == ERROR for f in findings) def test_dangling_reference_link_flagged(tmp_path): @@ -183,7 +181,6 @@ def test_author_caps_warned(): assert "author-caps" in _rules(findings) -def test_format_findings_renders(): +def test_findings_carry_rule_and_severity(): findings = lint_content(CLEAN.replace("name: my-skill", "name: BAD")) - out = format_findings(findings) - assert "name-format" in out + assert any(f.rule == "name-format" and f.severity == ERROR for f in findings) diff --git a/tools/skill_ledger.py b/tools/skill_ledger.py index 6a5e6c580c..3d70c19e93 100644 --- a/tools/skill_ledger.py +++ b/tools/skill_ledger.py @@ -1,23 +1,12 @@ """Per-mutation skill audit ledger + single-edit rollback. -Every skill mutation — any actor — appends one JSONL entry to -``~/.hermes/skills/.curator_ledger.jsonl`` with before/after file manifests -whose contents are stored content-addressed (sha256-deduped) under -``~/.hermes/.curator_backups/blobs/``. - -Design decisions (Teknium-approved): - - JSONL, not the state DB: durable, human-greppable, survives DB resets. - - Covers ALL actors (``curator`` / ``agent`` / ``user``). The curator - invariant (never hard-delete autonomously) applies only to autonomous - actors; user deletes stay hard-delete but are ledgered and recoverable via - ``hermes curator rollback ``. - - Per-file content-addressed blobs, not tarballs: a mutation usually touches - one file, and identical content dedupes to one blob. - -The ledger is TELEMETRY, NOT A GATE: a ledger failure must never block the -mutation it describes — every public write path swallows and logs. The one -exception is ``rollback_entry``, which FAILS CLOSED when its own pre-rollback -safety capture fails (consistent with agent/curator_backup.py). +Every skill mutation (any actor) appends one JSONL entry to +``~/.hermes/skills/.curator_ledger.jsonl`` with before/after file manifests whose +contents are stored content-addressed (sha256-deduped) under +``~/.hermes/.curator_backups/blobs/``. JSONL, not the state DB: durable, +human-greppable, survives DB resets. The ledger is TELEMETRY, NOT A GATE: every +public write path swallows and logs. The one exception is ``rollback_entry``, +which FAILS CLOSED when its own pre-rollback safety capture fails. """ from __future__ import annotations @@ -38,10 +27,6 @@ from hermes_constants import get_hermes_home logger = logging.getLogger(__name__) -ACTOR_CURATOR = "curator" -ACTOR_AGENT = "agent" -ACTOR_USER = "user" - # Snapshot-id shape used by agent.curator_backup (duplicated so the ledger can # read the newest skills.tar.gz without importing the backup stack). _BACKUP_ID_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}-\d{2}-\d{2}Z(-\d{2})?$") @@ -51,13 +36,12 @@ _ARCHIVE_TS_SUFFIX_RE = re.compile(r"^(.+)-\d{14}$") # have re-homed support files out of the tree first, so a disk-only capture # would make rollback restore a hollow skill. _PACKAGE_RESTORE_ACTIONS = frozenset({"delete", "archive", "purge"}) -_VALID_ACTORS = {ACTOR_CURATOR, ACTOR_AGENT, ACTOR_USER} +_VALID_ACTORS = {"curator", "agent", "user"} _NON_PACKAGE_TOPS = {".curator_backups", ".hub", ".archive"} # Explicit actor override: the CLI sets "user", the curator walk sets "curator". _actor_override: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar( - "skill_ledger_actor", default=None -) + "skill_ledger_actor", default=None) def set_ledger_actor(actor: Optional[str]) -> contextvars.Token: @@ -76,18 +60,13 @@ def derive_actor() -> str: return override try: from tools.skill_provenance import is_background_review - if is_background_review(): - return ACTOR_CURATOR + return "curator" except Exception: pass - return ACTOR_AGENT + return "agent" -# --------------------------------------------------------------------------- -# Paths + config gate -# --------------------------------------------------------------------------- - def ledger_path() -> Path: return get_hermes_home() / "skills" / ".curator_ledger.jsonl" @@ -101,11 +80,10 @@ def _skills_dir() -> Path: def ledger_enabled() -> bool: - """Config gate ``skills.ledger`` (default True). Lazy import so this - module stays importable without the CLI config layer.""" + """Config gate ``skills.ledger`` (default True); lazy import keeps the module + importable without the CLI config layer.""" try: from hermes_cli.config import cfg_get, load_config - return bool(cfg_get(load_config(), "skills", "ledger", default=True)) except Exception as e: # pragma: no cover — best-effort config read logger.debug("skill_ledger: config read failed (%s); defaulting on", e) @@ -133,9 +111,7 @@ def _is_within(root: Path, path: Path) -> bool: return False -# --------------------------------------------------------------------------- -# Blob store (content-addressed, deduped) -# --------------------------------------------------------------------------- +# --- Blob store (content-addressed, deduped) --------------------------------- def _store_blob(data: bytes) -> str: """Write *data* keyed by sha256 (existing blob left alone). Returns the hash.""" @@ -160,11 +136,7 @@ def read_blob(sha256: str) -> Optional[bytes]: return None -def snapshot_paths( - root: Optional[Path], - *, - complete_package: bool = False, -) -> List[Dict[str, str]]: +def snapshot_paths(root: Optional[Path], *, complete_package: bool = False) -> List[Dict[str, str]]: """Capture {path, sha256} for every file under *root*, storing each as a blob. Empty when root is None/missing. Raises on I/O failure — callers decide @@ -188,9 +160,7 @@ def snapshot_paths( return out -# --------------------------------------------------------------------------- -# Package-completeness fill from the newest curator backup -# --------------------------------------------------------------------------- +# --- Package-completeness fill from the newest curator backup ----------------- def _package_rel(root: Path) -> Optional[str]: """Relative POSIX path of a skill dir under ``skills/``; None when outside @@ -215,10 +185,8 @@ def _skill_md_parent(items: Optional[List[Dict[str, str]]]) -> Optional[Path]: def package_prefixes( - root: Optional[Path] = None, - skill: Optional[str] = None, - before: Optional[List[Dict[str, str]]] = None, -) -> List[str]: + root: Optional[Path] = None, skill: Optional[str] = None, + before: Optional[List[Dict[str, str]]] = None) -> List[str]: """Tar member prefixes that belong to this skill's package: its live location under ``skills/``, the package parent recorded in the before-state SKILL.md path (for rollback fills where *root* is gone), the bare skill @@ -247,8 +215,7 @@ def _latest_skills_tarball() -> Optional[Path]: except OSError: return None candidates = [ - child / "skills.tar.gz" - for child in children + child / "skills.tar.gz" for child in children if child.is_dir() and _BACKUP_ID_RE.match(child.name) and (child / "skills.tar.gz").is_file() ] if not candidates: @@ -288,21 +255,17 @@ def _read_package_files_from_latest_backup(prefixes: List[str]) -> Dict[str, byt def fill_snapshot_from_curator_backup( - root: Optional[Path], - existing: Optional[List[Dict[str, str]]] = None, - *, - skill: Optional[str] = None, -) -> List[Dict[str, str]]: + root: Optional[Path], existing: Optional[List[Dict[str, str]]] = None, *, + skill: Optional[str] = None) -> List[Dict[str, str]]: """Union missing skill-package files from the newest curator snapshot. Completeness fill, not a gate: failures are swallowed and *existing* is - returned unchanged; the backup only fills paths ABSENT from it. - - Filled files are addressed where the rollback must restore them: under - *root* when known (for purge that is ``.archive//``, NOT the live - tree), else under the live skills dir. Backup members carry a leading - package-dir segment, stripped when *root* already names the package. - Every fill target must stay under ``skills/`` and HERMES_HOME. + returned unchanged; the backup only fills paths ABSENT from it. Filled files + are addressed where the rollback must restore them: under *root* when known + (for purge that is ``.archive//``, NOT the live tree), else under the + live skills dir. Backup members carry a leading package-dir segment, stripped + when *root* already names the package. Every fill target must stay under + ``skills/`` and HERMES_HOME. """ out = list(existing or []) prefixes = package_prefixes(root, skill, out) @@ -342,18 +305,12 @@ def fill_snapshot_from_curator_backup( return out -# --------------------------------------------------------------------------- -# Append + read -# --------------------------------------------------------------------------- +# --- Append + read ------------------------------------------------------------ def append_entry( - action: str, - skill: str, - before: Optional[List[Dict[str, str]]] = None, - after: Optional[List[Dict[str, str]]] = None, - actor: Optional[str] = None, - evidence: Optional[Dict[str, Any]] = None, -) -> Optional[str]: + action: str, skill: str, before: Optional[List[Dict[str, str]]] = None, + after: Optional[List[Dict[str, str]]] = None, actor: Optional[str] = None, + evidence: Optional[Dict[str, Any]] = None) -> Optional[str]: """Append one ledger entry. Returns the entry id, or None when the ledger is disabled or the write failed (never raises).""" if not ledger_enabled(): @@ -367,8 +324,7 @@ def append_entry( "skill": skill, "evidence": evidence or {}, "before": before or [], - "after": after or [], - } + "after": after or []} path = ledger_path() path.parent.mkdir(parents=True, exist_ok=True) with open(path, "a", encoding="utf-8") as fh: @@ -380,14 +336,9 @@ def append_entry( def record_mutation( - action: str, - skill: str, - before_root: Optional[Path] = None, - before: Optional[List[Dict[str, str]]] = None, - after_root: Optional[Path] = None, - actor: Optional[str] = None, - evidence: Optional[Dict[str, Any]] = None, -) -> Optional[str]: + action: str, skill: str, before_root: Optional[Path] = None, + before: Optional[List[Dict[str, str]]] = None, after_root: Optional[Path] = None, + actor: Optional[str] = None, evidence: Optional[Dict[str, Any]] = None) -> Optional[str]: """One-stop hook for mutation call sites: capture after-state from *after_root* (pre-captured *before* list, or capture from *before_root*) and append. NEVER raises and never blocks the mutation. @@ -404,18 +355,14 @@ def record_mutation( before = fill_snapshot_from_curator_backup(before_root, before, skill=skill) after = snapshot_paths(after_root) return append_entry( - action, skill, before=before, after=after, actor=actor, evidence=evidence - ) + action, skill, before=before, after=after, actor=actor, evidence=evidence) except Exception as e: logger.warning("skill_ledger: record_mutation failed (%s) — mutation unaffected", e) return None def capture_before( - root: Optional[Path], - *, - complete_package: bool = False, - skill: Optional[str] = None, + root: Optional[Path], *, complete_package: bool = False, skill: Optional[str] = None, ) -> Optional[List[Dict[str, str]]]: """Best-effort pre-mutation capture; None on failure or when disabled (callers pass the result straight to record_mutation). Use @@ -432,9 +379,7 @@ def capture_before( return None -def list_entries( - skill: Optional[str] = None, limit: Optional[int] = None -) -> List[Dict[str, Any]]: +def list_entries(skill: Optional[str] = None, limit: Optional[int] = None) -> List[Dict[str, Any]]: """Read the ledger, newest first. Malformed lines are skipped.""" path = ledger_path() if not path.exists(): @@ -468,9 +413,7 @@ def get_entry(entry_id: str) -> Optional[Dict[str, Any]]: return next((row for row in list_entries() if row.get("id") == entry_id), None) -# --------------------------------------------------------------------------- -# Single-edit rollback -# --------------------------------------------------------------------------- +# --- Single-edit rollback ----------------------------------------------------- def _validate_entry_paths(entry: Dict[str, Any]) -> Optional[str]: """All paths in an entry must live under HERMES_HOME — a hand-edited @@ -509,8 +452,7 @@ def rollback_entry(entry_id: str) -> Tuple[bool, str]: # added, and the filled set is re-validated against HERMES_HOME. if entry.get("action") in _PACKAGE_RESTORE_ACTIONS: before = fill_snapshot_from_curator_backup( - _skill_md_parent(before), before, skill=str(entry.get("skill") or "") or None - ) + _skill_md_parent(before), before, skill=str(entry.get("skill") or "") or None) path_err = _validate_entry_paths({**entry, "before": before, "after": after}) if path_err: return False, f"refusing rollback: {path_err}" @@ -518,10 +460,8 @@ def rollback_entry(entry_id: str) -> Tuple[bool, str]: # Pre-check every blob we need so we never fail mid-restore. for item in before: if read_blob(str(item.get("sha256", ""))) is None: - return False, ( - f"missing blob {item.get('sha256')} for {item.get('path')}; " - "rollback aborted, nothing was changed" - ) + return False, (f"missing blob {item.get('sha256')} for {item.get('path')}; " + "rollback aborted, nothing was changed") # Touched paths = union of before/after. Capture their CURRENT state as # the safety entry so the rollback itself is undoable. FAIL CLOSED. @@ -533,22 +473,14 @@ def rollback_entry(entry_id: str) -> Tuple[bool, str]: if fp.is_file(): safety_before.append({"path": p, "sha256": _store_blob(fp.read_bytes())}) safety_id = append_entry( - "pre-rollback", - entry.get("skill", "?"), - before=safety_before, - after=safety_before, - evidence={"rollback_target": entry_id}, - ) + "pre-rollback", entry.get("skill", "?"), before=safety_before, after=safety_before, + evidence={"rollback_target": entry_id}) except Exception as e: - return False, ( - f"pre-rollback safety capture failed ({e}); rollback aborted and " - "current skills were not changed" - ) + return False, (f"pre-rollback safety capture failed ({e}); rollback aborted and " + "current skills were not changed") if safety_id is None: - return False, ( - "pre-rollback safety capture failed (ledger disabled or " - "unwritable); rollback aborted and current skills were not changed" - ) + return False, ("pre-rollback safety capture failed (ledger disabled or " + "unwritable); rollback aborted and current skills were not changed") # Restore: write every before-file, remove files the mutation created. before_paths = {str(i["path"]) for i in before} @@ -573,14 +505,9 @@ def rollback_entry(entry_id: str) -> Tuple[bool, str]: logger.warning("skill_ledger: could not remove %s during rollback: %s", p, e) append_entry( - "rollback", - entry.get("skill", "?"), - before=safety_before, - after=before, - evidence={"rollback_target": entry_id, "restored": restored, "removed": removed}, - ) + "rollback", entry.get("skill", "?"), before=safety_before, after=before, + evidence={"rollback_target": entry_id, "restored": restored, "removed": removed}) return True, ( f"rolled back entry {entry_id} ({entry.get('action')} on " f"'{entry.get('skill')}'): {restored} file(s) restored, {removed} removed. " - f"Safety entry {safety_id} captured the pre-rollback state." - ) + f"Safety entry {safety_id} captured the pre-rollback state.") diff --git a/tools/skill_linter.py b/tools/skill_linter.py index 28a78abad8..37a30354fc 100644 --- a/tools/skill_linter.py +++ b/tools/skill_linter.py @@ -1,18 +1,11 @@ """Structural + convention linter for SKILL.md files. -The hard *validator* (``tools/skill_manager_tool.py::_validate_frontmatter``) -blocks the non-negotiables (fence, YAML mapping, name/description, size caps). -This module is the softer companion encoding the CONTRIBUTING.md "Skill -authoring standards" that otherwise only a human reviewer catches: shell -utilities named instead of native tools, missing author/license/metadata, -``name`` != directory, dangling ``references/`` links, marketing words, -``platforms:`` gating vs POSIX-only scripts, forbidden scaffolding files. - -Contract: findings are advisory — ``lint_skill`` returns ``LintFinding`` rows and -the caller decides what blocks. Pure functions (I/O limited to the files pointed -at) so CI, the ``skill_manage`` create path and local runs share one impl. -Frontmatter parsing is delegated to ``agent.skill_utils`` so BOM handling and -the prompt description budget stay in one place. +The hard validator (``skill_manager_tool._validate_frontmatter``) blocks the +non-negotiables; this is the advisory companion encoding the CONTRIBUTING.md +"Skill authoring standards" a human reviewer would otherwise catch. Findings +never block by themselves — ``lint_skill`` returns ``LintFinding`` rows and the +caller decides. Frontmatter parsing is delegated to ``agent.skill_utils`` so +BOM handling and the prompt description budget stay in one place. """ from __future__ import annotations @@ -22,39 +15,17 @@ from dataclasses import dataclass from pathlib import Path from typing import Any, Dict, List, Optional -from agent.skill_utils import ( - SKILL_PROMPT_DESC_LIMIT, - parse_frontmatter, -) - -# ── Rule data ──────────────────────────────────────────────────────────────── +from agent.skill_utils import SKILL_PROMPT_DESC_LIMIT, parse_frontmatter # Shell utilities already wrapped as native tools; naming them in prose steers # the model to a raw shell call. banned token -> native tool to name instead. _SHELL_UTIL_TO_TOOL: Dict[str, str] = { - "grep": "search_files", - "rg": "search_files", - "cat": "read_file", - "head": "read_file", - "tail": "read_file", - "sed": "patch", - "awk": "patch", - "find": "search_files (target='files')", - "ls": "search_files (target='files')", -} - -# Marketing words the description must not contain. + "grep": "search_files", "rg": "search_files", "cat": "read_file", "head": "read_file", + "tail": "read_file", "sed": "patch", "awk": "patch", + "find": "search_files (target='files')", "ls": "search_files (target='files')"} _MARKETING_WORDS = ( - "powerful", - "comprehensive", - "seamless", - "advanced", - "cutting-edge", - "state-of-the-art", - "revolutionary", - "robust", -) - + "powerful", "comprehensive", "seamless", "advanced", "cutting-edge", "state-of-the-art", + "revolutionary", "robust") # POSIX-only primitives that require ``platforms:`` when a bundled script uses # them. Detected in scripts/, not in prose. _POSIX_PRIMITIVES = ( @@ -65,19 +36,9 @@ _POSIX_PRIMITIVES = ( "osascript", "/proc/", "apt-get", - "systemctl", -) - + "systemctl") # Scaffolding files a skill should not ship (noise, not skill content). -_FORBIDDEN_FILES = ( - "README.md", - "CHANGELOG.md", - "install.sh", - ".env", - ".env.example", - ".gitignore", -) - +_FORBIDDEN_FILES = ("README.md", "CHANGELOG.md", "install.sh", ".env", ".env.example", ".gitignore") # Presence of the load-bearing section is checked, not exact ordering, so the # linter is not a change-detector. _EXPECTED_SECTIONS = ("When to Use", "When to use") @@ -94,10 +55,6 @@ class LintFinding: rule: str message: str - def format(self) -> str: - badge = "✗" if self.severity == ERROR else "⚠" - return f"{badge} [{self.rule}] {self.message}" - def _err(rule: str, message: str) -> LintFinding: return LintFinding(ERROR, rule, message) @@ -107,32 +64,23 @@ def _warn(rule: str, message: str) -> LintFinding: return LintFinding(WARNING, rule, message) -# ── Individual checks ──────────────────────────────────────────────────────── - - def _check_name_matches_dir( - frontmatter: Dict[str, Any], skill_dir: Optional[Path] + frontmatter: Dict[str, Any], skill_dir: Optional[Path], ) -> List[LintFinding]: if skill_dir is None: return [] name = str(frontmatter.get("name", "")).strip() if name and name != skill_dir.name: - return [_err( - "name-dir-mismatch", - f"frontmatter name '{name}' does not match directory " - f"'{skill_dir.name}'; they must be identical.", - )] + return [_err("name-dir-mismatch", f"frontmatter name '{name}' does not match directory " + f"'{skill_dir.name}'; they must be identical.")] return [] def _check_name_format(frontmatter: Dict[str, Any]) -> List[LintFinding]: name = str(frontmatter.get("name", "")).strip() if name and not re.fullmatch(r"[a-z0-9][a-z0-9_-]*", name): - return [_err( - "name-format", - f"name '{name}' must be lowercase letters, digits, hyphens, " - f"and underscores only.", - )] + return [_err("name-format", f"name '{name}' must be lowercase letters, digits, hyphens, " + f"and underscores only.")] return [] @@ -146,8 +94,8 @@ def _check_description(frontmatter: Dict[str, Any]) -> List[LintFinding]: if len(desc) > SKILL_PROMPT_DESC_LIMIT: findings.append(_warn( "description-length", - f"description is {len(desc)} chars; the skill index truncates " - f"past {SKILL_PROMPT_DESC_LIMIT} chars + '...', losing routing " + f"description is {len(desc)} chars; the skill index truncates past " + f"{SKILL_PROMPT_DESC_LIMIT} chars + '...', losing routing " f"signal. Keep it to one sentence.", )) lower = desc.lower() @@ -155,9 +103,7 @@ def _check_description(frontmatter: Dict[str, Any]) -> List[LintFinding]: if hits: findings.append(_warn( "description-marketing", - f"description contains marketing words {hits}; state the " - f"capability, not adjectives.", - )) + f"description contains marketing words {hits}; state the capability, not adjectives.")) return findings @@ -166,26 +112,20 @@ def _check_metadata_block(frontmatter: Dict[str, Any]) -> List[LintFinding]: for key in ("version", "author", "license"): if key not in frontmatter: findings.append(_warn( - "missing-metadata", - f"frontmatter is missing '{key}'; every peer skill has it.", - )) + "missing-metadata", f"frontmatter is missing '{key}'; every peer skill has it.")) meta = frontmatter.get("metadata") hermes_meta = meta.get("hermes") if isinstance(meta, dict) else None if not isinstance(hermes_meta, dict): findings.append(_warn( - "missing-metadata", - "frontmatter is missing metadata.hermes.{tags, related_skills}.", - )) + "missing-metadata", "frontmatter is missing metadata.hermes.{tags, related_skills}.")) elif "tags" not in hermes_meta: findings.append(_warn("missing-metadata", "metadata.hermes.tags is missing.")) author = str(frontmatter.get("author", "")) if author and author.strip().lower() in ("hermes", "agent", "hermes agent") and ( - author != "Hermes Agent" - ): + author != "Hermes Agent"): findings.append(_warn( "author-caps", - f"author '{author}' should be 'Hermes Agent' (proper caps) " - f"or a real contributor name.", + f"author '{author}' should be 'Hermes Agent' (proper caps) or a real contributor name.", )) return findings @@ -195,23 +135,16 @@ def _check_shell_utilities(body: str) -> List[LintFinding]: prose = _strip_code_blocks(body) # Only backtick-wrapped mentions: bare words in sentences are too noisy. return [ - _warn( - "shell-utility-reference", - f"prose references `{util}`; name the native tool " - f"`{tool}` instead.", - ) + _warn("shell-utility-reference", + f"prose references `{util}`; name the native tool `{tool}` instead.") for util, tool in _SHELL_UTIL_TO_TOOL.items() - if re.search(rf"`{re.escape(util)}`", prose) - ] + if re.search(rf"`{re.escape(util)}`", prose)] def _check_sections(body: str) -> List[LintFinding]: if not any(re.search(rf"^#+\s+{re.escape(s)}", body, re.M) for s in _EXPECTED_SECTIONS): - return [_warn( - "missing-section", - "no '## When to Use' section found; skills need explicit " - "trigger conditions near the top.", - )] + return [_warn("missing-section", "no '## When to Use' section found; skills need explicit " + "trigger conditions near the top.")] return [] @@ -231,16 +164,13 @@ def _check_reference_links(body: str, skill_dir: Optional[Path]) -> List[LintFin if "*" in rel or rel.endswith("/"): # placeholders / globs continue if not (skill_dir / rel).exists(): - findings.append(_warn( - "dangling-reference", - f"body references '{rel}' but that file does not exist " - f"in the skill directory.", - )) + findings.append(_warn("dangling-reference", f"body references '{rel}' but that file " + f"does not exist in the skill directory.")) return findings def _check_platforms_gating( - frontmatter: Dict[str, Any], skill_dir: Optional[Path] + frontmatter: Dict[str, Any], skill_dir: Optional[Path], ) -> List[LintFinding]: """If bundled scripts use POSIX-only primitives, require platforms:.""" if skill_dir is None or frontmatter.get("platforms"): @@ -263,10 +193,8 @@ def _check_platforms_gating( detail = "; ".join(f"{k}: {v}" for k, v in offenders.items()) return [_warn( "platforms-gating", - f"scripts use POSIX-only primitives ({detail}) but no " - f"'platforms:' frontmatter is declared. Fix cross-platform or " - f"gate with platforms: [linux, macos].", - )] + f"scripts use POSIX-only primitives ({detail}) but no 'platforms:' frontmatter is " + f"declared. Fix cross-platform or gate with platforms: [linux, macos].")] return [] @@ -274,14 +202,10 @@ def _check_forbidden_files(skill_dir: Optional[Path]) -> List[LintFinding]: if skill_dir is None: return [] return [ - _warn( - "forbidden-file", - f"skill ships '{fname}'; skills should not include " - f"scaffolding/config files.", - ) + _warn("forbidden-file", + f"skill ships '{fname}'; skills should not include scaffolding/config files.") for fname in _FORBIDDEN_FILES - if (skill_dir / fname).exists() - ] + if (skill_dir / fname).exists()] def _check_platform_list_valid(frontmatter: Dict[str, Any]) -> List[LintFinding]: @@ -292,11 +216,8 @@ def _check_platform_list_valid(frontmatter: Dict[str, Any]) -> List[LintFinding] items = platforms if isinstance(platforms, list) else [platforms] bad = [p for p in items if str(p).lower() not in valid] if bad: - return [_warn( - "platforms-value", - f"platforms contains unrecognized value(s) {bad}; expected a " - f"subset of {sorted(valid)}.", - )] + return [_warn("platforms-value", f"platforms contains unrecognized value(s) {bad}; " + f"expected a subset of {sorted(valid)}.")] return [] @@ -305,31 +226,25 @@ def _strip_code_blocks(body: str) -> str: return re.sub(r"```.*?```", "", body, flags=re.S) -# ── Public API ─────────────────────────────────────────────────────────────── - - -def lint_content( - content: str, *, skill_dir: Optional[Path] = None -) -> List[LintFinding]: +def lint_content(content: str, *, skill_dir: Optional[Path] = None) -> List[LintFinding]: """Lint raw SKILL.md *content*. - ``skill_dir`` enables on-disk checks (name/dir match, dangling links, - POSIX gating, forbidden files); without it only content checks run, - which is what the create path needs before the file exists. + ``skill_dir`` enables on-disk checks (name/dir match, dangling links, POSIX + gating, forbidden files); without it only content checks run, which is what + the create path needs before the file exists. """ frontmatter, body = parse_frontmatter(content) - findings: List[LintFinding] = [] - findings += _check_name_format(frontmatter) - findings += _check_name_matches_dir(frontmatter, skill_dir) - findings += _check_description(frontmatter) - findings += _check_metadata_block(frontmatter) - findings += _check_platform_list_valid(frontmatter) - findings += _check_shell_utilities(body) - findings += _check_sections(body) - findings += _check_reference_links(body, skill_dir) - findings += _check_platforms_gating(frontmatter, skill_dir) - findings += _check_forbidden_files(skill_dir) - return findings + return ( + _check_name_format(frontmatter) + + _check_name_matches_dir(frontmatter, skill_dir) + + _check_description(frontmatter) + + _check_metadata_block(frontmatter) + + _check_platform_list_valid(frontmatter) + + _check_shell_utilities(body) + + _check_sections(body) + + _check_reference_links(body, skill_dir) + + _check_platforms_gating(frontmatter, skill_dir) + + _check_forbidden_files(skill_dir)) def lint_skill(skill_md_path: Path) -> List[LintFinding]: @@ -337,60 +252,3 @@ def lint_skill(skill_md_path: Path) -> List[LintFinding]: skill_md_path = Path(skill_md_path) content = skill_md_path.read_text(encoding="utf-8", errors="ignore") return lint_content(content, skill_dir=skill_md_path.parent) - - -def format_findings(findings: List[LintFinding]) -> str: - """Render findings as a newline-joined human-readable block.""" - return "\n".join(f.format() for f in findings) - - -def has_errors(findings: List[LintFinding]) -> bool: - return any(f.severity == ERROR for f in findings) - - -def _main(argv: Optional[List[str]] = None) -> int: - """CLI: ``python -m tools.skill_linter ...`` - - Exit 1 only on ERROR-severity findings (WARNING-only exits 0) so CI can - gate on structural breakage without failing on advisory nits. - """ - import sys - - args = argv if argv is not None else sys.argv[1:] - if not args: - print("usage: python -m tools.skill_linter ...") - return 2 - - targets: List[Path] = [] - for arg in args: - p = Path(arg) - if p.is_dir(): - targets.extend(sorted(p.rglob("SKILL.md"))) - elif p.name == "SKILL.md" and p.is_file(): - targets.append(p) - else: - print(f"skip (not a SKILL.md or dir): {arg}") - - any_error = False - total = 0 - for skill_md in targets: - findings = lint_skill(skill_md) - if not findings: - continue - total += len(findings) - if has_errors(findings): - any_error = True - print(f"\n{skill_md.parent.name} ({skill_md}):") - print(format_findings(findings)) - - if total == 0: - print(f"All {len(targets)} skill(s) clean.") - else: - print(f"\n{total} finding(s) across {len(targets)} skill(s).") - return 1 if any_error else 0 - - -if __name__ == "__main__": - import sys - - sys.exit(_main()) diff --git a/tools/skill_manager_batch.py b/tools/skill_manager_batch.py index 7d436997b6..cb44296e39 100644 --- a/tools/skill_manager_batch.py +++ b/tools/skill_manager_batch.py @@ -19,9 +19,7 @@ _BATCH_MAX_OPS = 20 def _norm_target(op) -> str: fp = (op.get("file_path") or "").strip() - if not fp: - return "SKILL.md" - return posixpath.normpath(fp.lstrip("/")) + return posixpath.normpath(fp.lstrip("/")) if fp else "SKILL.md" def _validate_batch_ops(operations, default_name, tool_error): @@ -38,18 +36,15 @@ def _validate_batch_ops(operations, default_name, tool_error): f"operations[{i}]: unknown action '{act}'. " f"Batchable: {', '.join(sorted(_BATCH_OP_ACTIONS))}; " "delete must be sole.", - success=False, - ) + success=False) nm = op.get("name") or default_name if not nm: return None, tool_error(f"operations[{i}] needs a 'name' (the skill it targets).", success=False) names.append(nm) if act == "create" and nm in names[:-1]: return None, tool_error( - f"operations[{i}]: create for '{nm}' must precede that " - "skill's other ops.", - success=False, - ) + f"operations[{i}]: create for '{nm}' must precede that skill's other ops.", + success=False) preflight = _background_review_preflight(act, nm) if preflight is not None: return None, json.dumps(preflight, ensure_ascii=False) @@ -75,8 +70,7 @@ def _validate_batch_ops(operations, default_name, tool_error): "op would silently discard its work. One destructive op " "(write_file/remove_file/full rewrite) per file per batch; " "put it first, or fold the change in. Patch chains are fine.", - success=False, - ) + success=False) touched_files.add(key) return names, None @@ -147,29 +141,20 @@ def _rollback(snapshots, find_skill): except Exception as exc: # noqa: BLE001 notes.append( f"ROLLBACK FAILED for '{nm}' ({exc}); snapshot preserved at '{snap}'" - if snap is not None - else f"ROLLBACK FAILED for '{nm}' ({exc})" - ) + if snap is not None else f"ROLLBACK FAILED for '{nm}' ({exc})") return ("; ".join(notes) if notes else "all touched skills rolled back"), bool(notes) -def _skill_manage_batch( - operations, - default_name: str = None, - task_id: str = None, - session_id: str = None, -) -> str: +def _skill_manage_batch(operations, default_name: str = None, task_id: str = None, + session_id: str = None) -> str: """Apply a sequence of operations atomically (memory-tool pattern). - Each op carries its own ``name`` and ``action``. Every touched skill is - snapshotted before any op runs; any failure rolls ALL touched skills back - (skills the batch created are removed). - - Rules: ``delete`` only as the SOLE op (its recoverable-archive path doesn't - compose with rollback) and is routed to the single-op handler, preserving - absorbed_into/archive semantics; ``create`` must precede that skill's other - ops; the same-file clobber guard rejects silently-lost work. - ``default_name`` is the legacy top-level ``name`` fallback (staged replay). + Every touched skill is snapshotted before any op runs; any failure rolls ALL + touched skills back (skills the batch created are removed). ``delete`` is + only legal as the SOLE op (its recoverable-archive path doesn't compose with + rollback) and is routed to the single-op handler, preserving + absorbed_into/archive semantics. ``default_name`` is the legacy top-level + ``name`` fallback (staged replay). """ from tools import skill_manager_tool as _smt from tools.registry import tool_error @@ -183,19 +168,14 @@ def _skill_manage_batch( return tool_error( "delete must be the SOLE op in its call — it doesn't " "compose with other ops' rollback.", - success=False, - ) + success=False) op = operations[0] nm = op.get("name") or default_name if not nm: return tool_error("operations[0] (delete) needs a 'name'.", success=False) return _smt.skill_manage( - action="delete", - name=nm, - absorbed_into=op.get("absorbed_into"), - task_id=task_id, - session_id=session_id, - ) + action="delete", name=nm, absorbed_into=op.get("absorbed_into"), + task_id=task_id, session_id=session_id) names, err = _validate_batch_ops(operations, default_name, tool_error) if err is not None: @@ -220,8 +200,7 @@ def _skill_manage_batch( try: for i, op in enumerate(operations): raw = _smt._skill_manage_from( - {**op, "name": names[i]}, task_id=task_id, session_id=session_id, - ) + {**op, "name": names[i]}, task_id=task_id, session_id=session_id) try: parsed = json.loads(raw) except Exception: # noqa: BLE001 @@ -232,11 +211,9 @@ def _skill_manage_batch( "success": False, "error": ( f"operations[{i}] ({op['action']} on '{names[i]}') failed: " - f"{parsed.get('error', 'unknown error')} — batch aborted, {note}." - ), + f"{parsed.get('error', 'unknown error')} — batch aborted, {note}."), "failed_index": i, - "completed_before_failure": i, - } + "completed_before_failure": i} # Carry the failing op's teaching payload (patch's file_preview / # fuzzy-match hints) through — without it the model recovers blind. for k, v in parsed.items(): @@ -244,21 +221,15 @@ def _skill_manage_batch( fail.setdefault(k, v) return json.dumps(fail, ensure_ascii=False) results.append({"name": names[i], "action": op["action"], - "file_path": op.get("file_path"), - "success": True}) + "file_path": op.get("file_path"), "success": True}) finally: _smt._skill_gate_bypass.reset(token) if rollback_failed: # Keep the snapshots so the operator can still recover by hand. - logger.warning( - "skill_manage batch rollback failed, snapshots kept at %s", - snap_root, - ) + logger.warning("skill_manage batch rollback failed, snapshots kept at %s", snap_root) else: shutil.rmtree(snap_root, ignore_errors=True) return json.dumps( - {"success": True, "operations_applied": len(results), - "results": results}, - ensure_ascii=False, - ) + {"success": True, "operations_applied": len(results), "results": results}, + ensure_ascii=False) diff --git a/tools/skill_manager_guards.py b/tools/skill_manager_guards.py index f303249ce1..6c305f735b 100644 --- a/tools/skill_manager_guards.py +++ b/tools/skill_manager_guards.py @@ -14,13 +14,9 @@ from typing import Any, Dict, Optional logger = logging.getLogger("tools.skill_manager_tool") -_ERR_KEY = "error" - def _refusal(message: str, **extra: Any) -> Dict[str, Any]: - out: Dict[str, Any] = {"success": False, _ERR_KEY: message} - out.update(extra) - return out + return {"success": False, "error": message, **extra} def _is_background_review() -> bool: @@ -39,9 +35,7 @@ def _resolved_str(path: Path) -> str: return str(path) -# --------------------------------------------------------------------------- -# Background-review read marks -# --------------------------------------------------------------------------- +# --- Background-review read marks -------------------------------------------- class _BackgroundReviewReadMarks: """Read marks shared by copied tool contexts within one review run.""" @@ -59,18 +53,15 @@ class _BackgroundReviewReadMarks: return path in self._paths -_background_review_read_paths: ( - "_ctxvars.ContextVar[Optional[_BackgroundReviewReadMarks]]" -) = _ctxvars.ContextVar("background_review_read_paths", default=None) +_background_review_read_paths: "_ctxvars.ContextVar[Optional[_BackgroundReviewReadMarks]]" = ( + _ctxvars.ContextVar("background_review_read_paths", default=None)) def mark_background_review_skill_read(path: Path) -> None: """Record that the active background-review fork has read a skill file. - The review fork may evolve skills but must not patch content it only - inferred from the transcript: skill_view calls this after returning file - content, and the write guards below require the target to be marked. - """ + The fork must not patch content it only inferred from the transcript: + skill_view/read_file call this, and the write guards require the mark.""" if not _is_background_review(): return marks = _background_review_read_paths.get() @@ -90,9 +81,7 @@ def _reset_background_review_read_marks() -> None: _background_review_read_paths.set(_BackgroundReviewReadMarks()) -# --------------------------------------------------------------------------- -# Delete-target safety -# --------------------------------------------------------------------------- +# --- Delete-target safety ----------------------------------------------------- def _containing_skills_root(skill_path: Path) -> Path: """Skills root (local or external_dirs entry) containing ``skill_path``; @@ -123,20 +112,15 @@ def _is_path_redirect(path: Path) -> bool: def _validate_delete_target(skill_dir: Path) -> Optional[str]: - """Last-line guard before ``shutil.rmtree(skill_dir)``. - - ``_find_skill`` already restricts the dir to a real SKILL.md parent, but - even a poisoned tree must never recursively delete (1) a path outside every - known skills root, (2) a skills root itself, or (3) a symlink/junction - (rmtree would follow it). Returns a refusal string or ``None``. - """ + """Last-line guard before ``shutil.rmtree(skill_dir)``: even a poisoned tree + must never delete (1) a path outside every known skills root, (2) a skills + root itself, or (3) a symlink/junction (rmtree would follow it).""" from agent.skill_utils import get_all_skills_dirs if _is_path_redirect(skill_dir): return ( f"Refusing to delete '{skill_dir}': the skill directory is a " - f"symlink/junction. Remove the link target manually if intended." - ) + f"symlink/junction. Remove the link target manually if intended.") try: resolved = skill_dir.resolve() except OSError as exc: @@ -150,8 +134,7 @@ def _validate_delete_target(skill_dir: Path) -> Optional[str]: if resolved == root: return ( f"Refusing to delete '{skill_dir}': resolves to the skills root " - f"itself, which would remove every installed skill." - ) + f"itself, which would remove every installed skill.") try: rel = resolved.relative_to(root) except ValueError: @@ -159,31 +142,25 @@ def _validate_delete_target(skill_dir: Path) -> Optional[str]: if rel.parts: return None return ( - f"Refusing to delete '{skill_dir}': path does not resolve inside any " - f"known skills root." + f"Refusing to delete '{skill_dir}': path does not resolve inside any " f"known skills root." ) -# --------------------------------------------------------------------------- -# Ownership / provenance guards -# --------------------------------------------------------------------------- +# --- Ownership / provenance guards -------------------------------------------- def _pinned_guard(name: str) -> Optional[str]: """Refusal message if *name* is pinned or essential, else None. - Pin only guards against **deletion** (curator auto-archive and - ``skill_manage(delete)``); patches/edits stay allowed. Essential skills - (``ESSENTIAL_SKILLS``) are permanently pinned because the system prompt - references them. Best-effort: an unreadable sidecar lets the delete through. - """ + Pin only guards **deletion**; patches/edits stay allowed. ESSENTIAL_SKILLS are + permanently pinned (the system prompt references them). Best-effort: an + unreadable sidecar lets the delete through.""" try: from agent.skill_utils import ESSENTIAL_SKILLS if name in ESSENTIAL_SKILLS: return ( f"Skill '{name}' is essential to Hermes (the agent's own " f"operating manual referenced by the system prompt) and " - f"cannot be deleted. Patches and edits are still allowed." - ) + f"cannot be deleted. Patches and edits are still allowed.") except Exception: logger.debug("essential-guard lookup failed for %s", name, exc_info=True) try: @@ -194,24 +171,19 @@ def _pinned_guard(name: str) -> Optional[str]: f"skill_manage. Ask the user to run " f"`hermes curator unpin {name}` if they want to delete it. " f"Patches and edits are allowed on pinned skills; only " - f"deletion is blocked." - ) + f"deletion is blocked.") except Exception: logger.debug("pinned-guard lookup failed for %s", name, exc_info=True) return None def _background_review_write_guard( - name: str, - skill_dir: Path, - action: str, + name: str, skill_dir: Path, action: str, ) -> Optional[Dict[str, Any]]: """Refuse autonomous curator writes to anything but curator-owned sediment. - Foreground agents may edit external/bundled/hub skills at the user's - direction; the background review fork has no user in the loop, so it is - also blocked on pinned skills (stricter than ``_pinned_guard``). - """ + The background review fork has no user in the loop, so unlike foreground + agents it is also blocked on pinned/external/bundled/hub skills.""" if not _is_background_review(): return None @@ -222,8 +194,7 @@ def _background_review_write_guard( f"Refusing background curator {action} for pinned skill " f"'{name}': pinned skills are off-limits to autonomous " "maintenance. Ask the user to run " - f"`hermes curator unpin {name}` if they want it changed." - ) + f"`hermes curator unpin {name}` if they want it changed.") except Exception: logger.debug("pinned skill guard lookup failed for %s", name, exc_info=True) @@ -233,8 +204,7 @@ def _background_review_write_guard( return _refusal( f"Refusing background curator {action} for skill '{name}': " "the skill lives in skills.external_dirs, which are " - "externally owned and read-only to autonomous curation." - ) + "externally owned and read-only to autonomous curation.") except Exception: logger.debug("external skill guard lookup failed for %s", name, exc_info=True) @@ -243,13 +213,10 @@ def _background_review_write_guard( for predicate, label in ( (skill_usage.is_protected_builtin, "protected built-in"), (skill_usage.is_hub_installed, "hub-installed"), - (skill_usage.is_bundled, "bundled"), - ): + (skill_usage.is_bundled, "bundled")): if predicate(name): return _refusal( - f"Refusing background curator {action} for {label} " - f"skill '{name}'." - ) + f"Refusing background curator {action} for {label} " f"skill '{name}'.") # Not curator-managed (no `created_by: "agent"` marker) => user-owned. # A MISSING record and an explicit `created_by: null` must resolve # IDENTICALLY: keying on record presence made the policy depend on the @@ -258,32 +225,24 @@ def _background_review_write_guard( # both; `hermes curator adopt ` is the supported way in. usage_rec = skill_usage.load_usage().get(name) if not skill_usage._is_curator_managed_record(usage_rec): - if isinstance(usage_rec, dict): - _detail = f"created_by={usage_rec.get('created_by')!r}" - else: - _detail = "no usage record" + _detail = (f"created_by={usage_rec.get('created_by')!r}" if isinstance(usage_rec, dict) + else "no usage record") return _refusal( f"Refusing background curator {action} for skill " f"'{name}': the skill is not curator-managed ({_detail}). " "User-owned skills are off-limits to autonomous curation. " - f"Run `hermes curator adopt {name}` to opt it in." - ) + f"Run `hermes curator adopt {name}` to opt it in.") except Exception: logger.warning("owned skill guard lookup failed for %s", name, exc_info=True) return _refusal( f"Refusing background curator {action} for skill '{name}': " "agent ownership could not be verified because the provenance " - "record is unavailable or unreadable." - ) + "record is unavailable or unreadable.") return None def _background_review_read_before_write_guard( - name: str, - target: Path, - action: str, - file_label: str, -) -> Optional[Dict[str, Any]]: + name: str, target: Path, action: str, file_label: str) -> Optional[Dict[str, Any]]: """Require review forks to load the exact target before mutating it.""" if not _is_background_review() or _background_review_has_read(target): return None @@ -293,8 +252,7 @@ def _background_review_read_before_write_guard( "review turn. Call skill_view(name) for SKILL.md, or " "skill_view(name, file_path=...) for a supporting file, then " "retry the write using the content just returned.", - _read_before_write_required=True, - ) + _read_before_write_required=True) def _background_review_preflight(action: str, name: str) -> Optional[Dict[str, Any]]: @@ -309,18 +267,14 @@ def _background_review_preflight(action: str, name: str) -> Optional[Dict[str, A def _curator_consolidation_delete_guard( - name: str, absorbed_into: Optional[str] + name: str, absorbed_into: Optional[str], ) -> Optional[Dict[str, Any]]: """Fail closed on unverified deletes during the curator consolidation pass. - The review fork's only legitimate ``skill_manage(delete)`` is a verified - consolidation declared via ``absorbed_into=`` (existence is - validated in ``_delete_skill``). A bare delete (``None`` or ``""``) is the - fail-open behavior that once archived whole clusters of active skills; the - deterministic inactivity prune archives via ``skill_usage.archive_skill`` - without ever calling ``skill_manage``, so a bare prune here can only be the - LLM pass pruning without evidence. Refuse it; keep the skill active. - """ + The review fork's only legitimate delete is a verified consolidation declared + via ``absorbed_into=`` (existence validated in ``_delete_skill``). + The deterministic inactivity prune never calls ``skill_manage``, so a bare + delete here can only be the LLM pass pruning without evidence: refuse it.""" if not _is_background_review(): return None if isinstance(absorbed_into, str) and absorbed_into.strip(): @@ -333,20 +287,15 @@ def _curator_consolidation_delete_guard( "skill with no forwarding target is not permitted here — the " "deterministic inactivity prune handles staleness archival " "separately. Keeping '{name}' active.".format(name=name), - _fail_closed=True, - ) + _fail_closed=True) -# --------------------------------------------------------------------------- -# Org-mirror handling -# --------------------------------------------------------------------------- +# --- Org-mirror handling ------------------------------------------------------ def _maybe_auto_propose_org_edit(name: str, skill_path: Path) -> Optional[str]: """Submit an org-skill edit upstream when `sync.org_auto_propose` is on. - - Returns a short note for the tool result, or None when nothing happened. - Never raises: the edit is already saved locally and can be proposed later. - """ + Returns a note for the tool result or None; never raises (the edit is + already saved locally and can be proposed later).""" from tools import skill_manager_tool as _smt try: @@ -359,33 +308,27 @@ def _maybe_auto_propose_org_edit(name: str, skill_path: Path) -> Optional[str]: return ( f"This skill is shared by your organisation. Your edit is " f"saved locally and will not be overwritten by org updates. " - f"Run `hermes sync propose {name}` to share it back." - ) + f"Run `hermes sync propose {name}` to share it back.") result = ssc.propose_skill(name) if result.get("proposal_pending"): return ( f"Auto-proposed to your organisation as proposal " - f"#{result.get('proposal_id')} (pending admin review)." - ) + f"#{result.get('proposal_id')} (pending admin review).") return "Auto-proposed to your organisation (merged into the shared set)." except Exception as e: logger.debug("auto-propose skipped for %s: %s", name, e) return ( f"Edit saved locally. Could not submit it to your organisation " - f"right now — run `hermes sync propose {name}` to retry." - ) + f"right now — run `hermes sync propose {name}` to retry.") def _org_mirror_write_guard(name: str, skill_path: Path, action: str) -> Optional[Dict[str, Any]]: """Org-shared skills are EDITABLE IN PLACE — this only blocks deletion. - Refusing every write to `_org/` froze shared skills while personal ones - kept improving (agents don't fork mid-task). Edits now land in the mirror, - survive the next org pull (baseline sidecar in skills_sync_client), and - reach the org via `hermes sync propose` or `sync.org_auto_propose`. - Deletion stays refused: the mirror is a view of org HEAD, so a local - delete just comes back, and removing for everyone is an admin action. - """ + Edits land in the mirror, survive the next org pull (baseline sidecar in + skills_sync_client) and reach the org via `hermes sync propose`. Deletion + stays refused: the mirror is a view of org HEAD, so a local delete just + comes back, and removing for everyone is an admin action.""" if action not in {"delete", "remove_file"}: return None from tools import skill_manager_tool as _smt @@ -400,8 +343,7 @@ def _org_mirror_write_guard(name: str, skill_path: Path, action: str) -> Optiona "the next sync. Ask an org admin to remove it for " "everyone. (Editing it IS allowed — your changes are kept " "and can be proposed back with `hermes sync propose " - f"{name}`.)" - ) + f"{name}`.)") except Exception: logger.debug("org mirror guard lookup failed for %s", name, exc_info=True) return None diff --git a/tools/skill_manager_tool.py b/tools/skill_manager_tool.py index ae6d55a2cb..0e64b992c6 100644 --- a/tools/skill_manager_tool.py +++ b/tools/skill_manager_tool.py @@ -6,9 +6,7 @@ Lets the agent create, patch, and delete skills — its procedural memory (narrow, actionable "how to do X"), as opposed to MEMORY.md/USER.md (broad, declarative). New skills land in ~/.hermes/skills/ (or ``skills.create_dir``); existing skills (bundled, hub-installed, user-created) are modified in place. - -Actions: create, edit (legacy full rewrite), patch, delete, write_file, -remove_file. Layout: ``/[category/]/SKILL.md`` plus optional +Layout: ``/[category/]/SKILL.md`` plus optional ``references/ templates/ scripts/ assets/`` subdirs. """ @@ -30,8 +28,7 @@ from agent.skill_utils import ( extract_skill_description, is_skill_description_truncated_for_prompt, parse_frontmatter as _parse_frontmatter, - SKILL_PROMPT_DESC_LIMIT, -) + SKILL_PROMPT_DESC_LIMIT) from tools.skill_manager_guards import ( # noqa: F401 — re-exported for callers/tests _BackgroundReviewReadMarks, _background_review_has_read, @@ -48,24 +45,14 @@ from tools.skill_manager_guards import ( # noqa: F401 — re-exported for calle _reset_background_review_read_marks, _validate_delete_target, _is_background_review, - mark_background_review_skill_read, -) + mark_background_review_skill_read) from tools.skill_manager_batch import ( # noqa: F401 - _BATCH_MAX_OPS, - _BATCH_OP_ACTIONS, - _skill_manage_batch, + _BATCH_MAX_OPS, _BATCH_OP_ACTIONS, _skill_manage_batch, ) +from tools.skills_guard import scan_skill, should_allow_install, format_scan_report logger = logging.getLogger(__name__) -# External hub installs are always scanned; agent-created skills only when -# skills.guard_agent_created is on. -try: - from tools.skills_guard import scan_skill, should_allow_install, format_scan_report - _GUARD_AVAILABLE = True -except ImportError: - _GUARD_AVAILABLE = False - def _guard_agent_created_enabled() -> bool: """skills.guard_agent_created (default False): the agent can already run @@ -73,9 +60,7 @@ def _guard_agent_created_enabled() -> bool: try: from hermes_cli.config import load_config return is_truthy_value( - cfg_get(load_config(), "skills", "guard_agent_created"), - default=False, - ) + cfg_get(load_config(), "skills", "guard_agent_created"), default=False) except Exception: return False @@ -87,7 +72,7 @@ def _security_scan_skill(skill_dir: Path) -> Optional[str]: verdict means dangerous findings for an agent-created skill — surfaced as an error so the agent can retry with the flagged content removed. """ - if not _GUARD_AVAILABLE or not _guard_agent_created_enabled(): + if not _guard_agent_created_enabled(): return None try: result = scan_skill(skill_dir, source="agent-created") @@ -143,9 +128,7 @@ def _err(message: str) -> Dict[str, Any]: return {"success": False, "error": message} -# ============================================================================= -# Validation helpers -# ============================================================================= +# --- Validation helpers ------------------------------------------------------- def _validate_name(name: str) -> Optional[str]: """Validate a skill name. Returns error message or None if valid.""" @@ -154,10 +137,8 @@ def _validate_name(name: str) -> Optional[str]: if len(name) > MAX_NAME_LENGTH: return f"Skill name exceeds {MAX_NAME_LENGTH} characters." if not VALID_NAME_RE.match(name): - return ( - f"Invalid skill name '{name}'. Use lowercase letters, numbers, " - f"hyphens, dots, and underscores. Must start with a letter or digit." - ) + return (f"Invalid skill name '{name}'. Use lowercase letters, numbers, " + f"hyphens, dots, and underscores. Must start with a letter or digit.") return None @@ -170,10 +151,8 @@ def _validate_category(category: Optional[str]) -> Optional[str]: category = category.strip() if not category: return None - invalid = ( - f"Invalid category '{category}'. Use lowercase letters, numbers, " - "hyphens, dots, and underscores. Categories must be a single directory name." - ) + invalid = (f"Invalid category '{category}'. Use lowercase letters, numbers, " + "hyphens, dots, and underscores. Categories must be a single directory name.") if "/" in category or "\\" in category: return invalid if len(category) > MAX_NAME_LENGTH: @@ -215,8 +194,7 @@ def _validate_frontmatter(content: str, *, new_skill: bool = False) -> Optional[ f"{SKILL_PROMPT_DESC_LIMIT}-char system-prompt budget (one sentence, " f"trigger first, ends with a period). The skill index truncates " f"longer descriptions to {SKILL_PROMPT_DESC_LIMIT - 3} chars + '...', " - f"destroying the routing signal. Move detail into the skill body." - ) + f"destroying the routing signal. Move detail into the skill body.") if not content[end_match.end() + 3:].strip(): return "SKILL.md must have content after the frontmatter (instructions, procedures, etc.)." return None @@ -229,8 +207,7 @@ def _validate_content_size(content: str, label: str = "SKILL.md") -> Optional[st f"{label} content is {len(content):,} characters " f"(limit: {MAX_SKILL_CONTENT_CHARS:,}). " f"Consider splitting into a smaller SKILL.md with supporting files " - f"in references/ or templates/." - ) + f"in references/ or templates/.") return None @@ -288,8 +265,7 @@ def _find_skill(name: str) -> Optional[Dict[str, Any]]: except OSError: logger.debug( "skills dir resolve failed; categorized lookups fall back to the unresolved path", - exc_info=True, - ) + exc_info=True) local_root = _skills_dir() for skills_dir in get_all_skills_dirs(): @@ -309,11 +285,8 @@ def _find_skill(name: str) -> Optional[Dict[str, Any]]: def _find_skill_in_other_profiles(name: str) -> List[Tuple[str, Path]]: - """``(profile_name, skill_dir)`` pairs for OTHER profiles holding ``name``. - - Lets the "not found" error explain a wrong-profile mistake. Fail-quiet: - empty list when discovery fails. - """ + """``(profile_name, skill_dir)`` pairs for OTHER profiles holding ``name``; + lets the "not found" error explain a wrong-profile mistake. Fail-quiet.""" matches: List[Tuple[str, Path]] = [] try: from hermes_constants import get_default_hermes_root @@ -360,15 +333,13 @@ def _skill_not_found_error(name: str, suffix: str = "") -> str: f" A skill by that name exists in profile " f"'{other_profile}' ({other_path}). To edit it, switch " f"profiles (`hermes -p {other_profile}`) or edit the file " - f"directly (file tools / terminal)." - ) + f"directly (file tools / terminal).") elif others: names = ", ".join(f"'{p}'" for p, _ in others) base += ( f" Skills by that name exist in other profiles: {names}. " f"Switch profiles (`hermes -p `) to edit there, or " - f"edit the files directly (file tools / terminal)." - ) + f"edit the files directly (file tools / terminal).") else: base += " Use skills_list() to see available skills." return base + suffix @@ -396,21 +367,16 @@ def _validate_file_path(file_path: str) -> Optional[str]: return None -def _resolve_skill_target(skill_dir: Path, file_path: str) -> Tuple[Optional[Path], Optional[str]]: - """Resolve a supporting-file path and ensure it stays within the skill directory.""" +def _resolve_supporting_file(skill_dir: Path, file_path: str): + """Validate ``file_path`` and resolve it inside ``skill_dir``. + Returns (target, None) or (None, error_dict).""" from tools.path_security import validate_within_dir - target = skill_dir / file_path - error = validate_within_dir(target, skill_dir) - return (None, error) if error else (target, None) - - -def _resolve_supporting_file(skill_dir: Path, file_path: str): - """``_validate_file_path`` + ``_resolve_skill_target``. Returns (target, None) or (None, error_dict).""" err = _validate_file_path(file_path) if err: return None, _err(err) - target, err = _resolve_skill_target(skill_dir, file_path) + target = skill_dir / file_path + err = validate_within_dir(target, skill_dir) if err: return None, _err(err) return target, None @@ -418,17 +384,13 @@ def _resolve_supporting_file(skill_dir: Path, file_path: str): def _locate_for_write(name: str, action: str, not_found_suffix: str = ""): """Find the skill and run the org-mirror + background-review write guards. - - Returns ``(skill_dir, None)`` or ``(None, error_dict)``. - """ + Returns ``(skill_dir, None)`` or ``(None, error_dict)``.""" existing = _find_skill(name) if not existing: return None, _err(_skill_not_found_error(name, not_found_suffix)) skill_dir = existing["path"] - guard = ( - _org_mirror_write_guard(name, skill_dir, action) - or _background_review_write_guard(name, skill_dir, action) - ) + guard = (_org_mirror_write_guard(name, skill_dir, action) + or _background_review_write_guard(name, skill_dir, action)) if guard: return None, guard return skill_dir, None @@ -455,11 +417,6 @@ def _attach_org_note(result: Dict[str, Any], name: str, skill_dir: Path) -> None result["message"] = f"{result['message']} {org_note}" -# ============================================================================= -# Core actions -# ============================================================================= - - def _add_description_prompt_preview(result: Dict[str, Any], content: str) -> None: """Append a system_prompt_preview field when the description will be truncated.""" fm, _ = _parse_frontmatter(content) @@ -467,8 +424,7 @@ def _add_description_prompt_preview(result: Dict[str, Any], content: str) -> Non result["system_prompt_preview"] = ( f"System prompt will show: \"{extract_skill_description(fm)}\" — " f"keep the trigger self-contained in the first " - f"{SKILL_PROMPT_DESC_LIMIT - 3} chars." - ) + f"{SKILL_PROMPT_DESC_LIMIT - 3} chars.") def _attach_lint_findings(result: Dict[str, Any], skill_md: Path) -> None: @@ -476,31 +432,28 @@ def _attach_lint_findings(result: Dict[str, Any], skill_md: Path) -> None: hard rejects already ran in _validate_frontmatter).""" try: from tools.skill_linter import lint_skill # local import: optional path - findings = lint_skill(skill_md) except Exception: return if not findings: return result["lint_warnings"] = [ - {"severity": f.severity, "rule": f.rule, "message": f.message} - for f in findings - ] + {"severity": f.severity, "rule": f.rule, "message": f.message} for f in findings] result["lint_hint"] = ( "The skill was created. These are advisory authoring-convention " "findings (not blockers) — fix them with skill_manage(action='patch') " - "to match Hermes skill standards." - ) + "to match Hermes skill standards.") +# --- Core actions ------------------------------------------------------------- + def _create_skill(name: str, content: str, category: str = None) -> Dict[str, Any]: """Create a new user skill with SKILL.md content.""" err = ( _validate_name(name) or _validate_category(category) or _validate_frontmatter(content, new_skill=True) - or _validate_content_size(content) - ) + or _validate_content_size(content)) if err: return _err(err) existing = _find_skill(name) @@ -525,8 +478,7 @@ def _create_skill(name: str, content: str, category: str = None) -> Dict[str, An "message": f"Skill '{name}' created.", "path": _display_path, "skill_md": str(skill_md), - "_change": {"description": _description_preview(content)}, - } + "_change": {"description": _description_preview(content)}} if category: result["category"] = category result["hint"] = ( @@ -561,20 +513,14 @@ def _edit_skill(name: str, content: str) -> Dict[str, Any]: "success": True, "message": f"Skill '{name}' updated (full rewrite).", "path": str(skill_dir), - "_change": {"description": _description_preview(content)}, - } + "_change": {"description": _description_preview(content)}} _attach_org_note(result, name, skill_dir) _add_description_prompt_preview(result, content) return result -def _patch_skill( - name: str, - old_string: str, - new_string: str, - file_path: str = None, - replace_all: bool = False, -) -> Dict[str, Any]: +def _patch_skill(name: str, old_string: str, new_string: str, file_path: str = None, + replace_all: bool = False) -> Dict[str, Any]: """Targeted find-and-replace within SKILL.md (default) or a supporting file. Requires a unique match unless replace_all is True.""" if not old_string: @@ -585,8 +531,7 @@ def _patch_skill( "file. Read the target file first (read_file on the skill's SKILL.md, or the file " "named by file_path) and copy the snippet verbatim, then retry 'patch'. " "Do NOT fall back to action='write_file' — that rewrites the entire file and " - "destroys unrelated content." - ) + "destroys unrelated content.") if new_string is None: return _err("new_string is required for 'patch'. Use an empty string to delete matched text.") # No old_string == new_string guard here: fuzzy_find_and_replace rejects @@ -616,8 +561,7 @@ def _patch_skill( from tools.fuzzy_match import fuzzy_find_and_replace new_content, match_count, _strategy, match_error = fuzzy_find_and_replace( - content, old_string, new_string, replace_all - ) + content, old_string, new_string, replace_all) if match_error: try: from tools.fuzzy_match import format_no_match_hint @@ -643,9 +587,7 @@ def _patch_skill( "message": f"Patched {target_label} in skill '{name}' ({match_count} replacement{'s' if match_count > 1 else ''}).", "_change": { "old": old_string[:200] + ("…" if len(old_string) > 200 else ""), - "new": new_string[:200] + ("…" if len(new_string) > 200 else ""), - }, - } + "new": new_string[:200] + ("…" if len(new_string) > 200 else "")}} _attach_org_note(result, name, skill_dir) return result @@ -676,8 +618,7 @@ def _delete_skill(name: str, absorbed_into: Optional[str] = None) -> Dict[str, A if not _find_skill(absorbed_target): return _err( f"absorbed_into='{absorbed_target}' does not exist. " - f"Create or patch the umbrella skill first, then retry the delete." - ) + f"Create or patch the umbrella skill first, then retry the delete.") skills_root = _containing_skills_root(skill_dir) unsafe = _validate_delete_target(skill_dir) # defense-in-depth before rmtree @@ -699,8 +640,7 @@ def _delete_skill(name: str, absorbed_into: Optional[str] = None) -> Dict[str, A return { "success": True, "message": f"Skill '{name}' archived ({archive_msg}).{absorbed_note}", - "_archived": True, - } + "_archived": True} shutil.rmtree(skill_dir) _rmdir_if_empty(skill_dir.parent, skills_root) # empty category dir, never the root @@ -724,8 +664,7 @@ def _write_file(name: str, file_path: str, file_content: str) -> Dict[str, Any]: return _err( f"File content is {content_bytes:,} bytes " f"(limit: {MAX_SKILL_FILE_BYTES:,} bytes / 1 MiB). " - f"Consider splitting into smaller files." - ) + f"Consider splitting into smaller files.") err = _validate_content_size(file_content, label=file_path) if err: return _err(err) @@ -749,8 +688,7 @@ def _write_file(name: str, file_path: str, file_content: str) -> Dict[str, Any]: result = { "success": True, "message": f"File '{file_path}' written to skill '{name}'.", - "path": str(target), - } + "path": str(target)} _attach_org_note(result, name, skill_dir) return result @@ -777,13 +715,11 @@ def _remove_file(name: str, file_path: str) -> Dict[str, Any]: for subdir in ALLOWED_SUBDIRS if (skill_dir / subdir).exists() for f in (skill_dir / subdir).rglob("*") - if f.is_file() - ] + if f.is_file()] return { "success": False, "error": f"File '{file_path}' not found in skill '{name}'.", - "available_files": available if available else None, - } + "available_files": available if available else None} read_guard = _background_review_read_before_write_guard(name, target, "remove_file", file_path) if read_guard: return read_guard @@ -793,15 +729,12 @@ def _remove_file(name: str, file_path: str) -> Dict[str, Any]: return {"success": True, "message": f"File '{file_path}' removed from skill '{name}'."} -# ============================================================================= -# Main entry point -# ============================================================================= +# --- Main entry point --------------------------------------------------------- # Set while replaying an already-approved staged skill write so skill_manage() # does not re-gate (and re-stage) it. _skill_gate_bypass: "_ctxvars.ContextVar[bool]" = _ctxvars.ContextVar( - "skill_gate_bypass", default=False -) + "skill_gate_bypass", default=False) _GATED_ACTIONS = {"create", "edit", "patch", "delete", "write_file", "remove_file"} @@ -825,8 +758,7 @@ def _run_write_gate(build_staging): return json.dumps( {"success": True, "staged": True, "pending_id": record["id"], "gist": gist, "message": decision.message}, - ensure_ascii=False, - ) + ensure_ascii=False) def _apply_skill_write_gate(action, name, **payload_kwargs): @@ -843,15 +775,13 @@ def _apply_skill_write_gate(action, name, **payload_kwargs): content=payload_kwargs.get("content") or "", file_path=payload_kwargs.get("file_path") or "", old_string=payload_kwargs.get("old_string") or "", - new_string=payload_kwargs.get("new_string") or "", - ) + new_string=payload_kwargs.get("new_string") or "") return payload, gist return _run_write_gate(_staging) -_FLAT_OP_KEYS = ("content", "category", "file_path", "file_content", - "old_string", "new_string") +_FLAT_OP_KEYS = ("content", "category", "file_path", "file_content", "old_string", "new_string") def _skill_manage_from(payload: Dict[str, Any], **extra) -> str: @@ -861,8 +791,7 @@ def _skill_manage_from(payload: Dict[str, Any], **extra) -> str: name=payload.get("name", ""), replace_all=payload.get("replace_all", False), **{k: payload.get(k) for k in _FLAT_OP_KEYS}, - **extra, - ) + **extra) def apply_skill_pending(payload: Dict[str, Any]) -> str: @@ -871,10 +800,8 @@ def apply_skill_pending(payload: Dict[str, Any]) -> str: token = _skill_gate_bypass.set(True) try: return _skill_manage_from( - payload, - absorbed_into=payload.get("absorbed_into"), - operations=payload.get("operations"), - ) + payload, absorbed_into=payload.get("absorbed_into"), + operations=payload.get("operations")) finally: _skill_gate_bypass.reset(token) @@ -891,12 +818,11 @@ def _maybe_debounced_sync_push(skill_name: str) -> None: Fast-path: skills not opted into sync do nothing (no auth, no network). The push runs via ``skills_sync_client.maybe_push_skills`` which enforces - the access gate and swallows errors. Never blocks the caller (M1-C). + the access gate and swallows errors. Never blocks the caller. """ global _sync_push_timer, _sync_push_lock try: from tools.skill_usage import is_sync_enabled - if not is_sync_enabled(skill_name): return except Exception: @@ -908,7 +834,6 @@ def _maybe_debounced_sync_push(skill_name: str) -> None: def _fire(): try: from tools.skills_sync_client import maybe_push_skills - maybe_push_skills(message=f"sync: {skill_name}") except Exception: pass @@ -941,8 +866,7 @@ def _act_patch(a): return tool_error( "Pass EITHER content (full SKILL.md rewrite) OR " "old_string/new_string (targeted replacement), not both.", - success=False, - ) + success=False) if a["content"]: return _edit_skill(a["name"], a["content"]) # Targeted-replacement validation lives in _patch_skill so the public @@ -972,8 +896,7 @@ _ACTION_HANDLERS = { "patch": _act_patch, "delete": lambda a: _delete_skill(a["name"], absorbed_into=a["absorbed_into"]), "write_file": _act_write_file, - "remove_file": _act_remove_file, -} + "remove_file": _act_remove_file} def _record_success(action, name, result, *, file_path, absorbed_into, task_id, @@ -993,12 +916,10 @@ def _record_success(action, name, result, *, file_path, absorbed_into, task_id, if file_path: _evidence["file_path"] = file_path _ledger.record_mutation( - action, - name, + action, name, before=ledger_before if ledger_before is not None else [], after_root=_post["path"] if _post else None, - evidence=_evidence, - ) + evidence=_evidence) except Exception: pass try: @@ -1032,20 +953,10 @@ def _record_success(action, name, result, *, file_path, absorbed_into, task_id, def skill_manage( - action: str, - name: str, - content: str = None, - category: str = None, - file_path: str = None, - file_content: str = None, - old_string: str = None, - new_string: str = None, - replace_all: bool = False, - absorbed_into: str = None, - task_id: str = None, - session_id: str = None, - operations=None, -) -> str: + action: str, name: str, content: str = None, category: str = None, file_path: str = None, + file_content: str = None, old_string: str = None, new_string: str = None, + replace_all: bool = False, absorbed_into: str = None, task_id: str = None, + session_id: str = None, operations=None) -> str: """Manage user-created skills; dispatches to the action handler. ``operations``: batch shape — a list of {action, ...} dicts applied @@ -1054,9 +965,7 @@ def skill_manage( """ if operations is not None: return _skill_manage_batch( - operations, default_name=name or None, - task_id=task_id, session_id=session_id, - ) + operations, default_name=name or None, task_id=task_id, session_id=session_id) preflight = _background_review_preflight(action, name) if preflight is not None: return json.dumps(preflight, ensure_ascii=False) @@ -1067,8 +976,7 @@ def skill_manage( args = dict( content=content, category=category, file_path=file_path, file_content=file_content, old_string=old_string, new_string=new_string, - replace_all=replace_all, absorbed_into=absorbed_into, - ) + replace_all=replace_all, absorbed_into=absorbed_into) gate_result = _apply_skill_write_gate(action, name, **args) if gate_result is not None: return gate_result @@ -1082,10 +990,7 @@ def skill_manage( from tools import skill_ledger as _ledger _pre = _find_skill(name) _ledger_before = _ledger.capture_before( - _pre["path"] if _pre else None, - complete_package=(action == "delete"), - skill=name, - ) + _pre["path"] if _pre else None, complete_package=(action == "delete"), skill=name) except Exception: pass @@ -1100,14 +1005,11 @@ def skill_manage( if result.get("success"): _record_success( action, name, result, file_path=file_path, absorbed_into=absorbed_into, - task_id=task_id, session_id=session_id, ledger_before=_ledger_before, - ) + task_id=task_id, session_id=session_id, ledger_before=_ledger_before) return json.dumps(result, ensure_ascii=False) -# ============================================================================= -# OpenAI Function-Calling Schema -# ============================================================================= +# --- OpenAI Function-Calling Schema ------------------------------------------- SKILL_MANAGE_SCHEMA = { "name": "skill_manage", @@ -1128,8 +1030,7 @@ SKILL_MANAGE_SCHEMA = { "op only). Existing skills are modified wherever they live. Keep " "the description's first 57 chars a self-contained trigger: 'Use " "when . .' — skill_view() shows " - "format conventions." - ), + "format conventions."), "parameters": { "type": "object", "properties": { @@ -1144,25 +1045,19 @@ SKILL_MANAGE_SCHEMA = { "description": ( "Skill name (lowercase, hyphens/underscores, " "max 64 chars); an existing skill's name " - "unless creating." - ) - }, + "unless creating.")}, "action": { "type": "string", - "enum": ["create", "patch", "delete", "write_file", "remove_file"] - }, + "enum": ["create", "patch", "delete", "write_file", "remove_file"]}, "content": { "type": "string", "description": ( "Full SKILL.md text (YAML frontmatter + " "markdown body) for create, or a full " - "rewrite on patch." - ) - }, + "rewrite on patch.")}, "category": { "type": "string", - "description": "Optional category subdir for create (e.g. 'devops')." - }, + "description": "Optional category subdir for create (e.g. 'devops')."}, # patch args: same fuzzy-matching semantics as the # `patch` tool — teach only skill-specific facts here. "old_string": { @@ -1171,12 +1066,10 @@ SKILL_MANAGE_SCHEMA = { }, "new_string": { "type": "string", - "description": "Replacement (patch); empty string deletes the match." - }, + "description": "Replacement (patch); empty string deletes the match."}, "replace_all": { "type": "boolean", - "description": "patch: replace all occurrences (default false)." - }, + "description": "patch: replace all occurrences (default false)."}, "file_path": { "type": "string", "description": ( @@ -1185,17 +1078,10 @@ SKILL_MANAGE_SCHEMA = { "never absolute. write_file/remove_file: " "required; first segment references/, " "templates/, scripts/, or assets/. patch: " - "optional (default SKILL.md)." - ) - }, + "optional (default SKILL.md).")}, "file_content": { - "type": "string", - "description": "Content for write_file." - } - }, - "required": ["name", "action"] - } - }, + "type": "string", "description": "Content for write_file."}}, + "required": ["name", "action"]}}, # NOTE: the handler also accepts the legacy flat single-op shape # (top-level action/name/content/old_string/new_string/ # replace_all/category/file_path/file_content) — old transcripts @@ -1204,9 +1090,7 @@ SKILL_MANAGE_SCHEMA = { # documents it and the delete guard's error re-teaches it). # None are advertised. }, - "required": ["operations"], - }, -} + "required": ["operations"]}} # --- Registry --- @@ -1222,5 +1106,4 @@ registry.register( operations=args.get("operations"), task_id=kw.get("task_id"), session_id=kw.get("session_id")), - emoji="📝", -) + emoji="📝")