From b187d4e54751f43e1642081debcb5e117c50f397 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:43:10 -0700 Subject: [PATCH] refactor(tools): skill linter checks as generators, skill_manage arg-shape table, suppress() collapses --- tools/skill_ledger.py | 109 ++++++-------- tools/skill_linter.py | 270 +++++++++++++--------------------- tools/skill_manager_batch.py | 40 ++--- tools/skill_manager_guards.py | 31 ++-- tools/skill_manager_tool.py | 124 ++++++---------- 5 files changed, 212 insertions(+), 362 deletions(-) diff --git a/tools/skill_ledger.py b/tools/skill_ledger.py index 3d70c19e93..7307df4a89 100644 --- a/tools/skill_ledger.py +++ b/tools/skill_ledger.py @@ -13,6 +13,7 @@ from __future__ import annotations import contextvars import hashlib +from contextlib import suppress import json import logging import os @@ -58,12 +59,10 @@ def derive_actor() -> str: override = _actor_override.get() if override in _VALID_ACTORS: return override - try: + with suppress(Exception): from tools.skill_provenance import is_background_review if is_background_review(): return "curator" - except Exception: - pass return "agent" @@ -80,8 +79,7 @@ def _skills_dir() -> Path: def ledger_enabled() -> bool: - """Config gate ``skills.ledger`` (default True); lazy import keeps the module - importable without the CLI config layer.""" + """Config gate ``skills.ledger`` (default True); lazy import keeps this importable without the CLI.""" try: from hermes_cli.config import cfg_get, load_config return bool(cfg_get(load_config(), "skills", "ledger", default=True)) @@ -129,20 +127,19 @@ def read_blob(sha256: str) -> Optional[bytes]: """Return blob content or None when missing/invalid.""" if not sha256 or not all(c in "0123456789abcdef" for c in sha256): return None - p = blobs_dir() / sha256 try: + p = blobs_dir() / sha256 return p.read_bytes() if p.exists() else None except OSError: return None 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. + """{path, sha256} for every file under *root*, each stored as a blob. - Empty when root is None/missing. Raises on I/O failure — callers decide - whether that is fatal (rollback safety capture) or swallowed (telemetry). - ``complete_package=True`` unions in files from the newest curator - ``skills.tar.gz`` for this skill (disk hashes win).""" + Empty when root is None/missing. Raises on I/O failure — callers decide whether + that is fatal (rollback safety capture) or swallowed (telemetry). + ``complete_package`` unions in the newest curator tarball's files (disk hashes win).""" if root is None: return [] root = Path(root) @@ -151,7 +148,7 @@ def snapshot_paths(root: Optional[Path], *, complete_package: bool = False) -> L elif root.is_dir(): files = sorted(p for p in root.rglob("*") if p.is_file()) elif complete_package: - files = [] + files = [] # gone from disk; the backup fill below may still recover it else: return [] out = [{"path": str(f), "sha256": _store_blob(f.read_bytes())} for f in files] @@ -163,8 +160,8 @@ def snapshot_paths(root: Optional[Path], *, complete_package: bool = False) -> L # --- 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 - it or under backup/hub/archive metadata roots (never a package).""" + """Relative POSIX path of a skill dir under ``skills/``; None when outside it + or under backup/hub/archive metadata roots (never a package).""" posix = (_rel_posix(root, _skills_dir()) or "").strip("/") if not posix or posix.split("/", 1)[0] in _NON_PACKAGE_TOPS: return None @@ -187,10 +184,9 @@ 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]: - """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 - name, and the name minus an archive collision suffix.""" + """Tar member prefixes of this skill's package: live location under ``skills/``, + the package parent from the before-state SKILL.md path (rollback fills where + *root* is gone), the bare skill name, and the name minus an archive suffix.""" candidates = [_package_rel(Path(root)) if root is not None else None] for item in before or []: path = Path(str(item.get("path", ""))) @@ -208,25 +204,20 @@ def package_prefixes( def _latest_skills_tarball() -> Optional[Path]: """Newest ``skills.tar.gz`` under ``skills/.curator_backups/``.""" backups = _skills_dir() / ".curator_backups" - if not backups.is_dir(): - return None try: - children = list(backups.iterdir()) + children = list(backups.iterdir()) if backups.is_dir() else [] except OSError: return None candidates = [ 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: - return None + if child.is_dir() and _BACKUP_ID_RE.match(child.name) and (child / "skills.tar.gz").is_file()] # Parent dirs sort lexicographically == chronologically for the id shape. - return max(candidates, key=lambda p: p.parent.name) + return max(candidates, key=lambda p: p.parent.name) if candidates else None def _read_package_files_from_latest_backup(prefixes: List[str]) -> Dict[str, bytes]: - """``{posix-relpath: bytes}`` for files under *prefixes* in the newest - snapshot. Malicious member names (absolute, ``..`` traversal) are rejected.""" + """``{posix-relpath: bytes}`` under *prefixes* in the newest snapshot; malicious + member names (absolute, ``..`` traversal) are rejected.""" if not prefixes: return {} archive = _latest_skills_tarball() @@ -259,14 +250,12 @@ def fill_snapshot_from_curator_backup( 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. - """ + Completeness fill, not a gate: failures return *existing* unchanged, and only + ABSENT paths are filled. Fill targets go where rollback must restore them: + under *root* when known (for purge that is ``.archive//``, NOT the live + tree), else the live skills dir; the tar's leading package-dir segment is + stripped when *root* already names the package. Every target must stay under + ``skills/`` and HERMES_HOME.""" out = list(existing or []) prefixes = package_prefixes(root, skill, out) if not prefixes: @@ -311,8 +300,7 @@ 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]: - """Append one ledger entry. Returns the entry id, or None when the - ledger is disabled or the write failed (never raises).""" + """Append one entry -> id, or None when disabled / write failed (never raises).""" if not ledger_enabled(): return None try: @@ -339,12 +327,10 @@ 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]: - """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. - - delete/archive/purge always capture a COMPLETE package (support files - filled from the newest curator backup) so rollback never restores a shell.""" + """Mutation hook: after-state from *after_root* (before = pre-captured list or + captured from *before_root*), then append. NEVER raises. delete/archive/purge + capture a COMPLETE package (filled from the newest curator backup) so + rollback never restores a shell.""" if not ledger_enabled(): return None try: @@ -364,9 +350,8 @@ def record_mutation( def capture_before( 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 - ``complete_package=True`` for delete/archive/purge captures.""" + """Best-effort pre-mutation capture; None on failure/disabled (pass straight to + record_mutation). ``complete_package=True`` for delete/archive/purge.""" if not ledger_enabled(): return None try: @@ -388,11 +373,8 @@ def list_entries(skill: Optional[str] = None, limit: Optional[int] = None) -> Li try: with open(path, "r", encoding="utf-8") as fh: for line in fh: - line = line.strip() - if not line: - continue try: - row = json.loads(line) + row = json.loads(line) if line.strip() else None except json.JSONDecodeError: continue if isinstance(row, dict): @@ -416,8 +398,8 @@ def get_entry(entry_id: str) -> Optional[Dict[str, Any]]: # --- 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 - ledger must not become a write-anywhere primitive.""" + """Every entry path must be under HERMES_HOME — a hand-edited ledger must not + become a write-anywhere primitive.""" home = get_hermes_home() for section in ("before", "after"): for item in entry.get(section) or []: @@ -428,13 +410,10 @@ def _validate_entry_paths(entry: Dict[str, Any]) -> Optional[str]: def rollback_entry(entry_id: str) -> Tuple[bool, str]: - """Restore the before-state of the single mutation *entry_id*. - - Fail-closed (mirrors agent/curator_backup.rollback): - 1. Every needed before-blob must exist — verified BEFORE any change. - 2. A pre-rollback safety entry capturing the CURRENT state of every - touched path is appended first; if that fails, nothing is changed. - """ + """Restore the before-state of mutation *entry_id*. Fail-closed (mirrors + agent/curator_backup.rollback): every before-blob must exist BEFORE any + change, and a pre-rollback safety entry of every touched path's CURRENT + state is appended first — if that fails, nothing is changed.""" entry = get_entry(entry_id) if entry is None: return False, f"no ledger entry with id '{entry_id}'" @@ -446,10 +425,9 @@ def rollback_entry(entry_id: str) -> Tuple[bool, str]: before = list(entry.get("before") or []) after = list(entry.get("after") or []) - # Historical hollow delete/archive/purge entries (``files: 1`` = SKILL.md): - # fill the before-state from the newest curator backup so the rollback - # restores the complete package. Entry hashes win; only missing paths are - # added, and the filled set is re-validated against HERMES_HOME. + # Historical hollow delete/archive/purge entries (SKILL.md only): fill from the + # newest curator backup so the complete package is restored. Entry hashes win; + # only missing paths are added, and the filled set is re-validated. 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) @@ -463,8 +441,7 @@ def rollback_entry(entry_id: str) -> Tuple[bool, str]: 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. + # Safety entry: CURRENT state of every touched path, so the rollback itself is undoable. touched = {str(i["path"]) for i in before + after if i.get("path")} try: safety_before: List[Dict[str, str]] = [] diff --git a/tools/skill_linter.py b/tools/skill_linter.py index 37a30354fc..9692a175fe 100644 --- a/tools/skill_linter.py +++ b/tools/skill_linter.py @@ -13,7 +13,7 @@ from __future__ import annotations import re from dataclasses import dataclass from pathlib import Path -from typing import Any, Dict, List, Optional +from typing import Any, Dict, Iterator, List, Optional from agent.skill_utils import SKILL_PROMPT_DESC_LIMIT, parse_frontmatter @@ -64,168 +64,107 @@ def _warn(rule: str, message: str) -> LintFinding: return LintFinding(WARNING, rule, message) -def _check_name_matches_dir( - 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 [] - - -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 [] - - -def _check_description(frontmatter: Dict[str, Any]) -> List[LintFinding]: - findings: List[LintFinding] = [] - # Measure the raw authored value: extract_skill_description() already - # truncates to the prompt budget, so it can never exceed the limit. - desc = str(frontmatter.get("description", "")).strip().strip("'\"") - if not desc: - return findings - if len(desc) > SKILL_PROMPT_DESC_LIMIT: - findings.append(_warn( - "description-length", - 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() - hits = [w for w in _MARKETING_WORDS if re.search(rf"\b{re.escape(w)}\b", lower)] - if hits: - findings.append(_warn( - "description-marketing", - f"description contains marketing words {hits}; state the capability, not adjectives.")) - return findings - - -def _check_metadata_block(frontmatter: Dict[str, Any]) -> List[LintFinding]: - findings: 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.")) - 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}.")) - 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"): - findings.append(_warn( - "author-caps", - f"author '{author}' should be 'Hermes Agent' (proper caps) or a real contributor name.", - )) - return findings - - -def _check_shell_utilities(body: str) -> List[LintFinding]: - """Flag banned shell utilities named in PROSE (not fenced code blocks).""" - 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 `{tool}` instead.") - for util, tool in _SHELL_UTIL_TO_TOOL.items() - 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 [] - - -def _check_reference_links(body: str, skill_dir: Optional[Path]) -> List[LintFinding]: - """Flag references/ links in the body that don't resolve on disk.""" - if skill_dir is None: - return [] - findings: List[LintFinding] = [] - seen: set[str] = set() - # Only references/, templates/, assets/ are reliably skill-owned; `scripts/` - # is excluded because dev skills legitimately cite repo-root scripts. - for match in re.finditer(r"(references|templates|assets)/[\w./-]+", body): - rel = match.group(0) - if rel in seen: - continue - seen.add(rel) - 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 " - f"does not exist in the skill directory.")) - return findings - - -def _check_platforms_gating( - 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"): - return [] - scripts_dir = skill_dir / "scripts" - if not scripts_dir.is_dir(): - return [] - offenders: Dict[str, List[str]] = {} - for script in scripts_dir.rglob("*"): - if not script.is_file() or script.suffix not in (".py", ".sh", ".bash"): - continue - try: - text = script.read_text(encoding="utf-8", errors="ignore") - except OSError: - continue - hit = [p for p in _POSIX_PRIMITIVES if p in text] - if hit: - offenders[script.name] = hit - if offenders: - detail = "; ".join(f"{k}: {v}" for k, v in offenders.items()) - return [_warn( - "platforms-gating", - f"scripts use POSIX-only primitives ({detail}) but no 'platforms:' frontmatter is " - f"declared. Fix cross-platform or gate with platforms: [linux, macos].")] - return [] - - -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 scaffolding/config files.") - for fname in _FORBIDDEN_FILES - if (skill_dir / fname).exists()] - - -def _check_platform_list_valid(frontmatter: Dict[str, Any]) -> List[LintFinding]: - platforms = frontmatter.get("platforms") - if not platforms: - return [] - valid = {"linux", "macos", "windows", "darwin"} - 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}; " - f"expected a subset of {sorted(valid)}.")] - return [] - - def _strip_code_blocks(body: str) -> str: """Remove fenced code blocks so prose-only checks don't fire on examples.""" return re.sub(r"```.*?```", "", body, flags=re.S) +def _check_frontmatter(frontmatter: Dict[str, Any], skill_dir: Optional[Path]) -> Iterator[LintFinding]: + name = str(frontmatter.get("name", "")).strip() + if name and not re.fullmatch(r"[a-z0-9][a-z0-9_-]*", name): + yield _err("name-format", f"name '{name}' must be lowercase letters, digits, hyphens, " + f"and underscores only.") + if skill_dir is not None and name and name != skill_dir.name: + yield _err("name-dir-mismatch", f"frontmatter name '{name}' does not match directory " + f"'{skill_dir.name}'; they must be identical.") + # Measure the raw authored value: extract_skill_description() already + # truncates to the prompt budget, so it can never exceed the limit. + desc = str(frontmatter.get("description", "")).strip().strip("'\"") + if len(desc) > SKILL_PROMPT_DESC_LIMIT: + yield _warn("description-length", + 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.") + hits = [w for w in _MARKETING_WORDS if re.search(rf"\b{re.escape(w)}\b", desc.lower())] + if hits: + yield _warn("description-marketing", + f"description contains marketing words {hits}; state the capability, not adjectives.") + for key in ("version", "author", "license"): + if key not in frontmatter: + yield _warn("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): + yield _warn("missing-metadata", "frontmatter is missing metadata.hermes.{tags, related_skills}.") + elif "tags" not in hermes_meta: + yield _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" + ): + yield _warn("author-caps", f"author '{author}' should be 'Hermes Agent' (proper caps) " + f"or a real contributor name.") + platforms = frontmatter.get("platforms") + if platforms: + valid = {"linux", "macos", "windows", "darwin"} + items = platforms if isinstance(platforms, list) else [platforms] + bad = [p for p in items if str(p).lower() not in valid] + if bad: + yield _warn("platforms-value", f"platforms contains unrecognized value(s) {bad}; " + f"expected a subset of {sorted(valid)}.") + + +def _check_body(body: str, skill_dir: Optional[Path]) -> Iterator[LintFinding]: + # Only backtick-wrapped mentions in PROSE (not fenced code): bare words are too noisy. + prose = _strip_code_blocks(body) + for util, tool in _SHELL_UTIL_TO_TOOL.items(): + if re.search(rf"`{re.escape(util)}`", prose): + yield _warn("shell-utility-reference", + f"prose references `{util}`; name the native tool `{tool}` instead.") + if not any(re.search(rf"^#+\s+{re.escape(s)}", body, re.M) for s in _EXPECTED_SECTIONS): + yield _warn("missing-section", "no '## When to Use' section found; skills need explicit " + "trigger conditions near the top.") + if skill_dir is None: + return + # Dangling links. Only references/, templates/, assets/ are reliably skill-owned; + # `scripts/` is excluded because dev skills legitimately cite repo-root scripts. + seen: set[str] = set() + for match in re.finditer(r"(references|templates|assets)/[\w./-]+", body): + rel = match.group(0) + if rel in seen or "*" in rel or rel.endswith("/"): # dupes, placeholders, globs + continue + seen.add(rel) + if not (skill_dir / rel).exists(): + yield _warn("dangling-reference", f"body references '{rel}' but that file " + f"does not exist in the skill directory.") + + +def _check_files(frontmatter: Dict[str, Any], skill_dir: Path) -> Iterator[LintFinding]: + # Bundled scripts using POSIX-only primitives require a platforms: declaration. + scripts_dir = skill_dir / "scripts" + offenders: Dict[str, List[str]] = {} + if not frontmatter.get("platforms") and scripts_dir.is_dir(): + for script in scripts_dir.rglob("*"): + if not script.is_file() or script.suffix not in (".py", ".sh", ".bash"): + continue + try: + text = script.read_text(encoding="utf-8", errors="ignore") + except OSError: + continue + hit = [p for p in _POSIX_PRIMITIVES if p in text] + if hit: + offenders[script.name] = hit + if offenders: + detail = "; ".join(f"{k}: {v}" for k, v in offenders.items()) + yield _warn("platforms-gating", + f"scripts use POSIX-only primitives ({detail}) but no 'platforms:' frontmatter is " + f"declared. Fix cross-platform or gate with platforms: [linux, macos].") + for fname in _FORBIDDEN_FILES: + if (skill_dir / fname).exists(): + yield _warn("forbidden-file", + f"skill ships '{fname}'; skills should not include scaffolding/config files.") + + def lint_content(content: str, *, skill_dir: Optional[Path] = None) -> List[LintFinding]: """Lint raw SKILL.md *content*. @@ -234,17 +173,10 @@ def lint_content(content: str, *, skill_dir: Optional[Path] = None) -> List[Lint the create path needs before the file exists. """ frontmatter, body = parse_frontmatter(content) - 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)) + findings = list(_check_frontmatter(frontmatter, skill_dir)) + list(_check_body(body, skill_dir)) + if skill_dir is not None: + findings += _check_files(frontmatter, skill_dir) + return findings def lint_skill(skill_md_path: Path) -> List[LintFinding]: diff --git a/tools/skill_manager_batch.py b/tools/skill_manager_batch.py index cb44296e39..2e506692d1 100644 --- a/tools/skill_manager_batch.py +++ b/tools/skill_manager_batch.py @@ -1,8 +1,6 @@ -"""Atomic multi-op batch path for ``skill_manage`` (extracted from skill_manager_tool). - -``skill_manage``/``_find_skill``/``_skill_gate_bypass`` are reached lazily -through ``tools.skill_manager_tool`` so the origin module owns all state. -""" +"""Atomic multi-op batch path for ``skill_manage``. Origin state +(``skill_manage``/``_find_skill``/``_skill_gate_bypass``) is reached lazily +through ``tools.skill_manager_tool`` so that module owns it.""" import json import logging @@ -49,11 +47,9 @@ def _validate_batch_ops(operations, default_name, tool_error): if preflight is not None: return None, json.dumps(preflight, ensure_ascii=False) - # Intra-batch clobber guard: sequential last-wins would SILENTLY discard an - # earlier op's work. A DESTRUCTIVE op (create/write_file/remove_file/full - # SKILL.md rewrite) on a file an earlier op touched is rejected; additive - # patches are always legal, so patch chains and write-then-patch stay - # allowed. Paths are normalized so spelling variants can't slip past. + # Clobber guard: a DESTRUCTIVE op (create/write_file/remove_file/full rewrite) on + # a file an earlier op touched would SILENTLY discard its work — reject it. + # Additive patches are always legal. Paths are normalized against spelling variants. touched_files = set() for i, op in enumerate(operations): act = op["action"] @@ -108,18 +104,15 @@ def _snapshot_skills(names, snap_root, find_skill): def _restore_snapshot(pre_dir, snap, post_dir) -> None: if snap is not None: if post_dir is not None and post_dir.is_dir(): - # Never destroy the only other copy before the restore lands: move - # the broken state aside and delete it only after the snapshot is - # back, so a failed copytree (disk full, locked file) can't turn - # into total skill loss. + # Move the broken state aside and delete it only after the snapshot is + # back, so a failed copytree (disk full, locked file) can't mean total loss. aside = post_dir.with_name(post_dir.name + ".rollback-broken") shutil.rmtree(aside, ignore_errors=True) post_dir.rename(aside) try: shutil.copytree(snap, pre_dir) except Exception: - # Restore failed: put the broken (half-applied) state back - # rather than leaving nothing. + # Restore failed: put the half-applied state back rather than nothing. shutil.rmtree(pre_dir, ignore_errors=True) aside.rename(pre_dir) raise @@ -147,15 +140,11 @@ def _rollback(snapshots, find_skill): 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). - - Every touched skill is snapshotted before any op runs; any failure rolls ALL - touched skills back (skills the batch created are removed). ``delete`` is + """Apply operations atomically: every touched skill is snapshotted first and any + failure rolls ALL of them back (batch-created skills 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). - """ + rollback) and routes to the single-op handler. ``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 @@ -192,8 +181,7 @@ def _skill_manage_batch(operations, default_name: str = None, task_id: str = Non shutil.rmtree(snap_root, ignore_errors=True) return tool_error(snap_err, success=False) - # Execute through the single-op path with the gate bypassed (the batch - # already cleared/staged it); ledger + telemetry fire per-op. + # Single-op path with the gate bypassed (the batch already cleared/staged it). results = [] rollback_failed = False token = _smt._skill_gate_bypass.set(True) diff --git a/tools/skill_manager_guards.py b/tools/skill_manager_guards.py index 6c305f735b..713488e3f4 100644 --- a/tools/skill_manager_guards.py +++ b/tools/skill_manager_guards.py @@ -1,9 +1,8 @@ -"""Write/delete guards for ``skill_manage`` (extracted from skill_manager_tool). +"""Write/delete guards for ``skill_manage``. -Every guard returns ``None`` when the operation may proceed, otherwise a -refusal (error dict or message). Origin-owned state (``_find_skill``, -``_skills_dir``) is reached lazily through ``tools.skill_manager_tool`` so -test patches on that module keep working. +Every guard returns ``None`` when the operation may proceed, otherwise a refusal +(error dict or message). Origin-owned state (``_find_skill``, ``_skills_dir``) is +reached lazily through ``tools.skill_manager_tool`` so test patches keep working. """ import contextvars as _ctxvars @@ -84,8 +83,7 @@ def _reset_background_review_read_marks() -> None: # --- Delete-target safety ----------------------------------------------------- def _containing_skills_root(skill_path: Path) -> Path: - """Skills root (local or external_dirs entry) containing ``skill_path``; - falls back to the local skills dir when no root matches.""" + """Skills root (local or external_dirs) containing ``skill_path``; local dir if none match.""" from agent.skill_utils import get_all_skills_dirs from tools import skill_manager_tool as _smt @@ -103,8 +101,8 @@ def _containing_skills_root(skill_path: Path) -> Path: def _is_path_redirect(path: Path) -> bool: - """True when ``path`` is a symlink or (Windows 3.12+) a junction — either - lets a poisoned tree redirect ``shutil.rmtree`` outside the skills root.""" + """Symlink or (Windows 3.12+) junction — either lets a poisoned tree redirect + ``shutil.rmtree`` outside the skills root.""" try: return path.is_symlink() or (hasattr(path, "is_junction") and path.is_junction()) except OSError: @@ -217,12 +215,10 @@ def _background_review_write_guard( if predicate(name): return _refusal( 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 - # guard's own side effect (the first successful write created a null - # record and the next identical write was refused). Fail closed for - # both; `hermes curator adopt ` is the supported way in. + # Not curator-managed (no `created_by: "agent"`) => user-owned. A MISSING + # record and an explicit `created_by: null` must resolve IDENTICALLY (keying + # on presence made the policy depend on the guard's own side effect: the + # first write created a null record, the next identical write was refused). usage_rec = skill_usage.load_usage().get(name) if not skill_usage._is_curator_managed_record(usage_rec): _detail = (f"created_by={usage_rec.get('created_by')!r}" if isinstance(usage_rec, dict) @@ -259,11 +255,8 @@ def _background_review_preflight(action: str, name: str) -> Optional[Dict[str, A if action not in {"edit", "patch", "delete", "write_file", "remove_file"}: return None from tools import skill_manager_tool as _smt - existing = _smt._find_skill(name) - if not existing: - return None - return _background_review_write_guard(name, existing["path"], action) + return _background_review_write_guard(name, existing["path"], action) if existing else None def _curator_consolidation_delete_guard( diff --git a/tools/skill_manager_tool.py b/tools/skill_manager_tool.py index 32f2c64c8f..55da42deb5 100644 --- a/tools/skill_manager_tool.py +++ b/tools/skill_manager_tool.py @@ -10,6 +10,7 @@ place. Layout: ``/[category/]/SKILL.md`` + optional import contextvars as _ctxvars import json +from contextlib import suppress import logging import re import shutil @@ -187,13 +188,10 @@ def _validate_content_size(content: str, label: str = "SKILL.md") -> Optional[st def _description_preview(content: str) -> str: """First 120 chars of the frontmatter description; '' on any failure.""" - try: + with suppress(Exception): fm_end = _FRONTMATTER_END_RE.search(content[3:]) if fm_end: - parsed = yaml.safe_load(content[3:fm_end.start() + 3]) - return str(parsed.get("description", ""))[:120] - except Exception: - pass + return str(yaml.safe_load(content[3:fm_end.start() + 3]).get("description", ""))[:120] return "" @@ -270,11 +268,9 @@ def _find_skill_in_other_profiles(name: str) -> List[Tuple[str, Path]]: # Every profile's skills dir EXCEPT the active one (already searched). candidates: List[Tuple[str, Path]] = [("default", root / "skills")] profiles_root = root / "profiles" - try: + with suppress(OSError): if profiles_root.is_dir(): candidates += [(e.name, e / "skills") for e in profiles_root.iterdir() if e.is_dir()] - except OSError: - pass for profile_name, skills_dir in candidates: try: @@ -519,11 +515,9 @@ def _patch_skill(name: str, old_string: str, new_string: str, file_path: str = N new_content, match_count, _strategy, match_error = fuzzy_find_and_replace( content, old_string, new_string, replace_all) if match_error: - try: + with suppress(Exception): from tools.fuzzy_match import format_no_match_hint match_error += format_no_match_hint(match_error, match_count, old_string, content) - except Exception: - pass return _err(match_error) | {"file_preview": content[:500] + ("..." if len(content) > 500 else "")} err = _validate_content_size(new_content, label=target_label) @@ -755,11 +749,9 @@ def _maybe_debounced_sync_push(skill_name: str) -> None: _sync_push_lock = threading.Lock() def _fire(): - try: + with suppress(Exception): from tools.skills_sync_client import maybe_push_skills maybe_push_skills(message=f"sync: {skill_name}") - except Exception: - pass with _sync_push_lock: if _sync_push_timer is not None: @@ -769,19 +761,6 @@ def _maybe_debounced_sync_push(skill_name: str) -> None: _sync_push_timer.start() -def _act_create(a): - if not a["content"]: - return tool_error("content is required for 'create'. Provide the full SKILL.md text (frontmatter + body).", success=False) - return _create_skill(a["name"], a["content"], a["category"]) - - -def _act_edit(a): - # Legacy alias for a full rewrite (old transcripts/callers; not in the schema). - if not a["content"]: - return tool_error("content is required for a full rewrite. Provide the full updated SKILL.md text.", success=False) - return _edit_skill(a["name"], a["content"]) - - def _act_patch(a): # Two shapes: old_string/new_string = targeted replacement; # content (alone) = full SKILL.md rewrite (absorbs the old 'edit'). @@ -795,36 +774,35 @@ def _act_patch(a): return _patch_skill(a["name"], a["old_string"], a["new_string"], a["file_path"], a["replace_all"]) -def _act_write_file(a): - if not a["file_path"]: - return tool_error("file_path is required for 'write_file'. Example: 'references/api-guide.md'", success=False) - if a["file_content"] is None: - return tool_error("file_content is required for 'write_file'.", success=False) - return _write_file(a["name"], a["file_path"], a["file_content"]) - - -def _act_remove_file(a): - if not a["file_path"]: - return tool_error("file_path is required for 'remove_file'.", success=False) - return _remove_file(a["name"], a["file_path"]) - - # action -> handler(args dict). Handlers return a result dict, or a JSON string -# (tool_error) for argument-shape errors. +# (tool_error) for argument-shape errors. "edit" is a legacy alias for a full +# rewrite (old transcripts/callers; not in the schema). _ACTION_HANDLERS = { - "create": _act_create, - "edit": _act_edit, + "create": lambda a: _create_skill(a["name"], a["content"], a["category"]), + "edit": lambda a: _edit_skill(a["name"], a["content"]), "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} + "write_file": lambda a: _write_file(a["name"], a["file_path"], a["file_content"]), + "remove_file": lambda a: _remove_file(a["name"], a["file_path"]), +} +# action -> (arg, is_missing, error) argument-shape checks run before the handler. +_REQUIRED_ARGS = { + "create": [("content", lambda v: not v, + "content is required for 'create'. Provide the full SKILL.md text (frontmatter + body).")], + "edit": [("content", lambda v: not v, + "content is required for a full rewrite. Provide the full updated SKILL.md text.")], + "write_file": [("file_path", lambda v: not v, + "file_path is required for 'write_file'. Example: 'references/api-guide.md'"), + ("file_content", lambda v: v is None, "file_content is required for 'write_file'.")], + "remove_file": [("file_path", lambda v: not v, "file_path is required for 'remove_file'.")], +} def _record_success(action, name, result, *, file_path, absorbed_into, task_id, session_id, ledger_before) -> None: """Best-effort post-mutation side effects (never break the tool): audit ledger, prompt-cache clear, curator telemetry, debounced sync push.""" - try: + with suppress(Exception): from tools import skill_ledger as _ledger _post = _find_skill(name) _evidence = {} @@ -839,17 +817,13 @@ def _record_success(action, name, result, *, file_path, absorbed_into, task_id, _ledger.record_mutation( action, name, before=ledger_before if ledger_before is not None else [], after_root=_post["path"] if _post else None, evidence=_evidence) - except Exception: - pass - try: + with suppress(Exception): from agent.prompt_builder import clear_skills_system_prompt_cache clear_skills_system_prompt_cache(clear_snapshot=True) - except Exception: - pass # Curator telemetry: only the background review fork marks a skill agent-created # (foreground creates belong to the user). A recoverable curator archive keeps its # record as STATE_ARCHIVED (`hermes curator status`/`restore`); only a hard delete forgets. - try: + with suppress(Exception): from tools.skill_usage import bump_patch, forget, record_created from tools.skill_provenance import is_background_review if action == "create": @@ -859,14 +833,10 @@ def _record_success(action, name, result, *, file_path, absorbed_into, task_id, bump_patch(name, action=action, task_id=task_id, session_id=session_id) elif action == "delete" and not result.get("_archived"): forget(name) - except Exception: - pass # Runs only AFTER the write gate passed (staged writes returned early), so # un-reviewed content is never pushed. - try: + with suppress(Exception): _maybe_debounced_sync_push(name) - except Exception: - pass def skill_manage( @@ -897,18 +867,19 @@ def skill_manage( # mutation. delete destroys the whole package (and consolidation may have re-homed # support files first), so complete it from the newest curator backup or a restore is hollow. _ledger_before = None - try: + with suppress(Exception): 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) - except Exception: - pass handler = _ACTION_HANDLERS.get(action) if handler is None: result = _err(f"Unknown action '{action}'. Use: create, edit, patch, delete, write_file, remove_file") else: + for arg, missing, message in _REQUIRED_ARGS.get(action, ()): + if missing(args[arg]): + return tool_error(message, success=False) result = handler({"name": name, **args}) if isinstance(result, str): return result # tool_error JSON for argument-shape problems @@ -924,11 +895,9 @@ def skill_manage( SKILL_MANAGE_SCHEMA = { "name": "skill_manage", - # ONE call shape (memory-tool pattern, maintainer-directed): the call - # IS an operations array — each op names its skill and action; a - # single edit is a list of one. The legacy flat shape (top-level - # action/name/content/...) is still ACCEPTED by the handler for old - # transcripts and staged-write replay, but no longer advertised. + # ONE advertised call shape (memory-tool pattern): the call IS an operations + # array. The legacy flat shape (top-level action/name/content/...) is still + # ACCEPTED for old transcripts and staged-write replay, but not advertised. "description": ( "Create, update, or delete skills — your procedural memory for " "recurring task types. The call is an operations array (a single " @@ -993,13 +962,9 @@ SKILL_MANAGE_SCHEMA = { "file_content": { "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 - # and staged-write replay depend on it — plus `absorbed_into` on - # delete ops (curator-only vocabulary; the curator's prompt - # documents it and the delete guard's error re-teaches it). - # None are advertised. + # Also accepted, never advertised: the legacy flat single-op fields, and + # `absorbed_into` on delete ops (curator-only vocabulary; the curator's + # prompt documents it and the delete guard's error re-teaches it). }, "required": ["operations"]}} @@ -1008,13 +973,8 @@ SKILL_MANAGE_SCHEMA = { from tools.registry import registry, tool_error registry.register( - name="skill_manage", - toolset="skills", - schema=SKILL_MANAGE_SCHEMA, + name="skill_manage", toolset="skills", schema=SKILL_MANAGE_SCHEMA, emoji="📝", handler=lambda args, **kw: _skill_manage_from( - args, - absorbed_into=args.get("absorbed_into"), - operations=args.get("operations"), - task_id=kw.get("task_id"), - session_id=kw.get("session_id")), - emoji="📝") + args, absorbed_into=args.get("absorbed_into"), operations=args.get("operations"), + task_id=kw.get("task_id"), session_id=kw.get("session_id")), +)