refactor(tools): skills group pass 2 — skill_usage on utils.atomic_write_text, hub-lock/lifecycle/relocate flattening; skills_guard dead full_content_hash out, ignore/trust/cache compaction; ast_audit + skillevaluator collapsed

tools/skill_provenance.py

@@ -1,18 +1,10 @@
-"""Skill write-origin provenance: a ContextVar separating agent-sediment skill writes from foreground
-user-directed writes. The curator only curates skills the background self-improvement review fork created;
-skills a user asked for belong to the user. run_agent.py binds the origin before each tool loop (mirroring
-AIAgent._memory_write_origin) so handlers such as skill_manage create can check it::
-
-    token = set_current_write_origin("background_review")
-    try:
-        ...  # tool runs here
-    finally:
-        reset_current_write_origin(token)
-"""
+"""Skill write-origin provenance: a ContextVar separating background-review skill writes from foreground
+user-directed writes (the curator only curates skills the self-improvement review fork created; skills a user
+asked for belong to the user). run_agent.py binds the origin before each tool loop, mirroring
+AIAgent._memory_write_origin: ``token = set_current_write_origin(...)`` / ``reset_current_write_origin(token)``."""
 
 import contextvars
 
-
 _write_origin: contextvars.ContextVar[str] = contextvars.ContextVar("skill_write_origin", default="foreground")
 
 # Sentinel used by the background review fork (run_agent._spawn_background_review).

tools/skill_usage.py

@@ -1,18 +1,14 @@
 """Skill usage telemetry + provenance for the Curator: a sidecar ``~/.hermes/skills/.usage.json`` keyed by
 skill name (never frontmatter — keeps telemetry out of user-authored SKILL.md and off bundled/hub skills).
-
-Counter bumps are best-effort (failures log at DEBUG, never break the tool call); writes are atomic
-(tempfile + os.replace) under a cross-process lock. Curator management is an explicit ``created_by: agent``
-marker written by skill_manage — never inferred from location. Lifecycle: active -> stale -> archived
-(moved to .archive/); ``pinned`` opts out of auto transitions, orthogonal to state.
-"""
+Counter bumps are best-effort (DEBUG-logged failures never break the tool call); writes are atomic under a
+cross-process lock. Curator management is an explicit ``created_by: agent`` marker written by skill_manage —
+never inferred from location. Lifecycle: active -> stale -> archived (moved to .archive/); ``pinned`` opts
+out of auto transitions, orthogonal to state."""
 
 from __future__ import annotations
 
 import json
 import logging
-import os
-import tempfile
 from contextlib import contextmanager, suppress
 from datetime import datetime, timezone
 from pathlib import Path
@@ -20,6 +16,7 @@ from typing import Any, Callable, Dict, Iterable, Iterator, List, Optional, Set,
 
 from hermes_constants import get_hermes_home
 from agent.skill_utils import is_excluded_skill_path, is_external_skill_path
+from utils import atomic_write_text
 
 logger = logging.getLogger(__name__)
 
@@ -29,10 +26,8 @@ try:
     import fcntl
 except ImportError:  # pragma: no cover - platform-specific fallback
     fcntl = None
-    try:
+    with suppress(ImportError):
         import msvcrt
-    except ImportError:
-        pass
 
 
 STATE_ACTIVE = "active"
@@ -41,8 +36,7 @@ STATE_ARCHIVED = "archived"
 _VALID_STATES = {STATE_ACTIVE, STATE_STALE, STATE_ARCHIVED}
 
 # Load-bearing built-ins (by frontmatter ``name``) the curator must NEVER archive/consolidate regardless of
-# ``curator.prune_builtins``, pins or LLM judgment — archiving one turns its slash command into "Unknown
-# command". Keep tiny; not a substitute for ``prune_builtins: false``. (``plan`` moved to a built-in command.)
+# ``curator.prune_builtins``, pins or LLM judgment — archiving one breaks its slash command. Keep tiny.
 PROTECTED_BUILTIN_SKILLS: Set[str] = set()
 
 
@@ -65,10 +59,9 @@ def _archive_dir() -> Path:
 
 def _flock(fd, lock: bool) -> None:
     if fcntl:
-        fcntl.flock(fd, fcntl.LOCK_EX if lock else fcntl.LOCK_UN)
-    else:
-        fd.seek(0)
-        msvcrt.locking(fd.fileno(), msvcrt.LK_LOCK if lock else msvcrt.LK_UNLCK, 1)
+        return fcntl.flock(fd, fcntl.LOCK_EX if lock else fcntl.LOCK_UN)
+    fd.seek(0)
+    msvcrt.locking(fd.fileno(), msvcrt.LK_LOCK if lock else msvcrt.LK_UNLCK, 1)
 
 
 @contextmanager
@@ -80,31 +73,14 @@ def _usage_file_lock():
         yield
         return
     if msvcrt and (not lock_path.exists() or lock_path.stat().st_size == 0):
-        lock_path.write_text(" ", encoding="utf-8")
-    fd = open(lock_path, "r+" if msvcrt else "a+", encoding="utf-8")
-    try:
-        _flock(fd, True)
-        yield
-    finally:
-        with suppress(OSError, IOError):
-            _flock(fd, False)
-        fd.close()
-
-
-def _atomic_write(path: Path, prefix: str, write: Callable[[Any], None]) -> None:
-    """Write *path* via tempfile + fsync + os.replace; the temp file is removed on failure."""
-    path.parent.mkdir(parents=True, exist_ok=True)
-    fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=prefix, suffix=".tmp")
-    try:
-        with os.fdopen(fd, "w", encoding="utf-8") as f:
-            write(f)
-            f.flush()
-            os.fsync(f.fileno())
-        os.replace(tmp, path)
-    except BaseException:
-        with suppress(OSError):
-            os.unlink(tmp)
-        raise
+        lock_path.write_text(" ", encoding="utf-8")  # msvcrt needs a non-empty byte range to lock
+    with open(lock_path, "r+" if msvcrt else "a+", encoding="utf-8") as fd:
+        try:
+            _flock(fd, True)
+            yield
+        finally:
+            with suppress(OSError, IOError):
+                _flock(fd, False)
 
 
 def _read_lines(path: Path, fail_log: str) -> List[str]:
@@ -112,11 +88,10 @@ def _read_lines(path: Path, fail_log: str) -> List[str]:
     if not path.exists():
         return []
     try:
-        lines = path.read_text(encoding="utf-8").splitlines()
+        return [s for s in (line.strip() for line in path.read_text(encoding="utf-8").splitlines()) if s]
     except OSError as e:
         logger.debug(fail_log, e)
         return []
-    return [s for s in (line.strip() for line in lines) if s]
 
 
 def _now_iso() -> str:
@@ -128,12 +103,11 @@ def _parse_iso_timestamp(value: Any) -> Optional[datetime]:
         parsed = datetime.fromisoformat(str(value)) if value else None
     except (TypeError, ValueError):
         return None
-    return parsed.replace(tzinfo=timezone.utc) if parsed is not None and parsed.tzinfo is None else parsed
+    return parsed.replace(tzinfo=timezone.utc) if parsed and parsed.tzinfo is None else parsed
 
 
 def latest_activity_at(record: Dict[str, Any]) -> Optional[str]:
-    """Newest use/view/patch timestamp. Creation time is excluded so never-active skills stay
-    distinguishable; lifecycle code falls back to ``created_at`` itself."""
+    """Newest use/view/patch timestamp; ``created_at`` is excluded so never-active skills stay distinguishable."""
     stamps = [(dt, str(raw)) for raw in (record.get(k) for k in ("last_used_at", "last_viewed_at", "last_patched_at"))
               if (dt := _parse_iso_timestamp(raw)) is not None]
     return max(stamps, key=lambda t: t[0])[1] if stamps else None
@@ -159,53 +133,45 @@ def _read_bundled_manifest_names() -> Set[str]:
 
 
 def _read_hub_installed_names() -> Set[str]:
-    """Names installed via the Skills Hub (``.hub/lock.json``, see tools/skills_hub.py::HubLockFile), plus the
-    frontmatter name of each ``install_path`` that resolves inside the skills dir."""
-    lock_path = _skills_dir() / ".hub" / "lock.json"
+    """Hub-installed names (``.hub/lock.json``) plus the frontmatter name of each in-tree ``install_path``."""
+    skills_dir = _skills_dir()
+    lock_path = skills_dir / ".hub" / "lock.json"
     if not lock_path.exists():
         return set()
     try:
         # errors="replace": hub descriptions can carry Windows-1252 high bytes; a strict read raises
         # UnicodeDecodeError (a ValueError, not caught below) and would 500 the whole /api/skills endpoint.
         data = json.loads(lock_path.read_text(encoding="utf-8", errors="replace"))
-        installed = (data.get("installed") or {}) if isinstance(data, dict) else None
-        if not isinstance(installed, dict):
-            return set()
-        names = {str(k) for k in installed}
-        skills_dir = _skills_dir()
-        for install_path in (e.get("install_path") for e in installed.values() if isinstance(e, dict)):
-            if not isinstance(install_path, str) or not install_path.strip():
-                continue
-            try:
-                resolved = (skills_dir / install_path).resolve()
-                resolved.relative_to(skills_dir.resolve())
-            except (OSError, ValueError):
-                continue
-            if (resolved / "SKILL.md").exists():
-                names.add(_read_skill_name(resolved / "SKILL.md", fallback=resolved.name))
-        return names
     except (OSError, json.JSONDecodeError) as e:
         logger.debug("Failed to read hub lock file: %s", e)
-    return set()
+        return set()
+    installed = (data.get("installed") or {}) if isinstance(data, dict) else None
+    if not isinstance(installed, dict):
+        return set()
+    names = {str(k) for k in installed}
+    paths = (e.get("install_path") for e in installed.values() if isinstance(e, dict))
+    for install_path in (p for p in paths if isinstance(p, str) and p.strip()):
+        with suppress(OSError, ValueError):  # ValueError: install_path escapes the skills dir
+            resolved = (skills_dir / install_path).resolve()
+            resolved.relative_to(skills_dir.resolve())
+            if (resolved / "SKILL.md").exists():
+                names.add(_read_skill_name(resolved / "SKILL.md", fallback=resolved.name))
+    return names
 
 
 def _prune_builtins_enabled() -> bool:
-    """``curator.prune_builtins`` (default True). Lazy config import keeps this module importable in the
-    update/sync context. The real mass-prune safety is seed-on-first-sight, not this flag."""
+    """``curator.prune_builtins`` (default True); lazy config import keeps this module importable during update/sync."""
     try:
         from hermes_cli.config import load_config
-        cfg = load_config()
-        cur = cfg.get("curator") if isinstance(cfg, dict) else None
-        if isinstance(cur, dict):
-            return bool(cur.get("prune_builtins", True))
+        cur = load_config().get("curator")
+        return bool(cur.get("prune_builtins", True)) if isinstance(cur, dict) else True
     except Exception as e:  # pragma: no cover — best-effort config read
         logger.debug("Failed to read curator.prune_builtins: %s", e)
-    return True
+        return True
 
 
 def read_suppressed_names() -> Set[str]:
-    """Built-ins the curator pruned (one name per line in ``.curator_suppressed``); the update-time re-seeder
-    must leave these archived, otherwise ``hermes update`` would re-copy the bundled skill."""
+    """Built-ins the curator pruned (``.curator_suppressed``); the update-time re-seeder must leave these archived."""
     lines = _read_lines(_skills_dir() / ".curator_suppressed", "Failed to read curator suppression list: %s")
     return {line for line in lines if not line.startswith("#")}
 
@@ -217,22 +183,21 @@ def _toggle_suppressed_name(skill_name: str, *, add: bool) -> None:
     (names.add if add else names.discard)(skill_name)
     data = "\n".join(sorted(names)) + ("\n" if names else "")
     try:
-        _atomic_write(_skills_dir() / ".curator_suppressed", ".curator_suppressed_", lambda f: f.write(data))
+        atomic_write_text(_skills_dir() / ".curator_suppressed", data, tmp_prefix=".curator_suppressed_")
     except Exception as e:
         logger.debug("Failed to write curator suppression list: %s", e, exc_info=True)
 
 
 def _iter_skill_mds(base: Path, *, local_only: bool) -> Iterator[Tuple[str, Path]]:
     """``(frontmatter name, SKILL.md)`` for flat and nested skills under *base*, skipping metadata/VCS/venv/cache
-    dirs. *local_only* also skips external skill dirs mounted below the tree (discovery sees them; curation must not)."""
+    dirs; *local_only* also skips external skill dirs mounted below the tree (curation must not touch them)."""
     for skill_md in base.rglob("SKILL.md"):
         if not (is_excluded_skill_path(skill_md) or (local_only and is_external_skill_path(skill_md))):
             yield _read_skill_name(skill_md, fallback=skill_md.parent.name), skill_md
 
 
 def _scan_local_skills(keep: Callable[[str, Path, Set[str], Dict[str, Any]], bool]) -> List[str]:
-    """Sorted unique names of local skills passing *keep(name, skill_md, bundled, usage)*. Hub-installed and
-    protected built-ins never reach *keep*."""
+    """Sorted local skill names passing *keep(name, skill_md, bundled, usage)*; hub/protected names never reach it."""
     base = _skills_dir()
     if not base.exists():
         return []
@@ -243,18 +208,17 @@ def _scan_local_skills(keep: Callable[[str, Path, Set[str], Dict[str, Any]], boo
 
 
 def list_agent_created_skill_names() -> List[str]:
-    """Skills the curator may manage: agent-authored (``created_by: agent`` record) plus, when
-    ``curator.prune_builtins`` is on, bundled built-ins (inactivity anchored on first sight). Never hub skills."""
+    """Curator-manageable skills: ``created_by: agent`` records plus, with ``curator.prune_builtins``, bundled
+    built-ins (which never carry a managed record, so the record gate applies only to local skills). Never hub."""
     prune_builtins = _prune_builtins_enabled()  # read once, before the walk
-    # Built-ins never carry a curator-managed record, so the record gate applies only to local skills.
     return _scan_local_skills(
         lambda name, _md, bundled, usage: prune_builtins if name in bundled else _is_curator_managed_record(usage.get(name)))
 
 
 def list_archived_skill_names() -> List[str]:
     """Skills in ``.archive/`` — flat layout (``archive_skill`` flattens), so dir name == skill name."""
-    archive_root = _archive_dir()
-    return sorted({p.name for p in archive_root.iterdir() if p.is_dir()}) if archive_root.exists() else []
+    root = _archive_dir()
+    return sorted({p.name for p in root.iterdir() if p.is_dir()}) if root.exists() else []
 
 
 def _read_skill_name(skill_md: Path, fallback: str) -> str:
@@ -274,9 +238,8 @@ def _read_skill_name(skill_md: Path, fallback: str) -> str:
 
 def is_agent_created(skill_name: str) -> bool:
     """Neither bundled nor hub-installed (and not only present in an external dir)."""
-    if skill_name in _read_bundled_manifest_names() | _read_hub_installed_names():
-        return False
-    return _find_skill_dir(skill_name) is not None or _find_external_skill_dir(skill_name) is None
+    return not (is_bundled(skill_name) or is_hub_installed(skill_name)) and (
+        _find_skill_dir(skill_name) is not None or _find_external_skill_dir(skill_name) is None)
 
 
 def is_hub_installed(skill_name: str) -> bool:
@@ -300,12 +263,12 @@ def is_curation_eligible(skill_name: str, skill_path: Optional[Path] = None) ->
     if is_bundled(skill_name):
         return _prune_builtins_enabled()
     local_dir = _find_skill_dir(skill_name)
-    return not is_external_skill_path(local_dir) if local_dir is not None else _find_external_skill_dir(skill_name) is None
+    return not is_external_skill_path(local_dir) if local_dir else _find_external_skill_dir(skill_name) is None
 
 
 def _is_curator_managed_record(record: Any) -> bool:
-    """``created_by`` reads like provenance but is a curator-management OPT-IN flag: ``"agent"`` means "curator-managed",
-    not proof of authorship (``hermes curator adopt`` flips it). Name kept: it is in every user's ``.usage.json``."""
+    """``created_by`` is a curator-management OPT-IN flag, not proof of authorship (``curator adopt`` flips it);
+    the key name is kept because it lives in every user's ``.usage.json``."""
     return isinstance(record, dict) and (record.get("created_by") == "agent" or record.get("agent_created") is True)
 
 
@@ -315,26 +278,25 @@ def is_curator_managed(skill_name: str) -> bool:
 
 
 def list_unmanaged_skill_names() -> List[str]:
-    """Curation-ELIGIBLE skills with no provenance marker (records predating ``created_by``, or FOREGROUND
-    ``skill_manage`` creates — those belong to the user). Invisible to ``curated_report()`` and automatic
-    transitions; surfaced by ``hermes curator status``, handed over only by explicit ``hermes curator adopt`` —
-    provenance is a declaration, never inferred from activity."""
+    """Curation-ELIGIBLE skills without a provenance marker (pre-``created_by`` records, or foreground creates that
+    belong to the user). Invisible to ``curated_report()`` and auto transitions; only ``curator adopt`` hands
+    them over — provenance is declared, never inferred from activity."""
     return _scan_local_skills(
         lambda name, md, bundled, usage: name not in bundled and not _is_curator_managed_record(usage.get(name))
         and is_curation_eligible(name, md))
 
 
 def unmanaged_report() -> List[Dict[str, Any]]:
-    """Rows for :func:`list_unmanaged_skill_names`. ``has_provenance_key`` False = no ``created_by`` key (pre-dates
-    the mechanism), True = present but unset (foreground create); explains WHY, not a signal to adopt on."""
+    """Rows for :func:`list_unmanaged_skill_names`; ``has_provenance_key`` (False = pre-dates ``created_by``, True =
+    present but unset) explains WHY, it is not a signal to adopt on."""
     usage = load_usage()
     return [_report_row(n, usage.get(n), has_provenance_key="created_by" in usage.get(n, {}), has_record=n in usage)
             for n in list_unmanaged_skill_names()]
 
 
 def adopt_skill(skill_name: str) -> Tuple[bool, str]:
-    """User-declared handover: writes the same ``created_by: agent`` marker the review fork writes; the inactivity
-    clock is NOT reset. Refuses hub, external, bundled and protected skills. Returns (ok, message)."""
+    """User-declared handover: writes the ``created_by: agent`` marker (inactivity clock NOT reset). Refuses hub,
+    external, bundled and protected skills. Returns (ok, message)."""
     if not skill_name:
         return False, "no skill name given"
     if is_protected_builtin(skill_name):
@@ -350,20 +312,19 @@ def adopt_skill(skill_name: str) -> Tuple[bool, str]:
         return False, f"skill '{skill_name}' not found"
     if is_external_skill_path(skill_dir):
         return False, _external_read_only_message(skill_name)
-    if _is_curator_managed_record(load_usage().get(skill_name)):
+    if is_curator_managed(skill_name):
         return True, f"'{skill_name}' is already curator-managed"
     mark_agent_created(skill_name)
-    if not _is_curator_managed_record(load_usage().get(skill_name)):
+    if not is_curator_managed(skill_name):
         return False, f"could not mark '{skill_name}' as curator-managed"
     return True, f"adopted '{skill_name}' into curator management"
 
 
 # --- Sidecar I/O ---
 def _empty_record() -> Dict[str, Any]:
-    return {
-        "created_by": None, "use_count": 0, "view_count": 0, "last_used_at": None, "last_viewed_at": None,
-        "patch_count": 0, "patch_generation": 0, "last_reused_patch_generation": 0, "last_patched_at": None,
-        "created_at": _now_iso(), "state": STATE_ACTIVE, "pinned": False, "archived_at": None}
+    return {"created_by": None, "use_count": 0, "view_count": 0, "last_used_at": None, "last_viewed_at": None,
+            "patch_count": 0, "patch_generation": 0, "last_reused_patch_generation": 0, "last_patched_at": None,
+            "created_at": _now_iso(), "state": STATE_ACTIVE, "pinned": False, "archived_at": None}
 
 
 def _backfilled(rec: Any) -> Dict[str, Any]:
@@ -384,10 +345,8 @@ def _report_row(name: str, raw: Any, **extra: Any) -> Dict[str, Any]:
 def load_usage() -> Dict[str, Dict[str, Any]]:
     """The whole .usage.json map (non-dict values dropped); {} on missing/corrupt."""
     path = _usage_file()
-    if not path.exists():
-        return {}
     try:
-        data = json.loads(path.read_text(encoding="utf-8"))
+        data = json.loads(path.read_text(encoding="utf-8")) if path.exists() else {}
         return {str(k): v for k, v in data.items() if isinstance(v, dict)} if isinstance(data, dict) else {}
     except (OSError, json.JSONDecodeError) as e:
         logger.debug("Failed to read %s: %s", path, e)
@@ -398,7 +357,7 @@ def save_usage(data: Dict[str, Dict[str, Any]]) -> bool:
     """Write the usage map atomically; True when it committed."""
     path = _usage_file()
     try:
-        _atomic_write(path, ".usage_", lambda f: json.dump(data, f, indent=2, sort_keys=True, ensure_ascii=False))
+        atomic_write_text(path, json.dumps(data, indent=2, sort_keys=True, ensure_ascii=False), tmp_prefix=".usage_")
         return True
     except Exception as e:
         logger.debug("Failed to write %s: %s", path, e, exc_info=True)
@@ -428,22 +387,18 @@ def _locked_update(
 
 
 def seed_record_if_missing(skill_name: str) -> None:
-    """Persist a baseline record for a curation-eligible skill so its inactivity clock is anchored at first
-    sight (``created_at`` = now), not at epoch. No-op if a record exists or the skill isn't eligible."""
-    if not skill_name or not is_curation_eligible(skill_name):
-        return
-
-    def _seed(data):  # load_usage() already dropped non-dict values, so "missing" == key absent
-        if missing := skill_name not in data:
-            data[skill_name] = _empty_record()
-        return None, missing
-    _locked_update(skill_name, _seed, "skill_usage.seed_record_if_missing(%s) failed: %s")
+    """Baseline record for a curation-eligible skill so its inactivity clock starts at first sight, not epoch."""
+    if skill_name and is_curation_eligible(skill_name):
+        # load_usage() already dropped non-dict values, so "missing" == key absent; dirty only when inserted.
+        def _seed(data):
+            return None, skill_name not in data and data.setdefault(skill_name, _empty_record()) is not None
+        _locked_update(skill_name, _seed, "skill_usage.seed_record_if_missing(%s) failed: %s")
 
 
 def _mutate(skill_name: str, mutator, *, require_curation_eligible: bool = False) -> Any:
-    """Load, apply *mutator(record)* in place, save; returns the mutator result (None if nothing landed).
-    Telemetry is recorded for ANY skill (observability is orthogonal to curation); lifecycle mutators pass
-    ``require_curation_eligible=True`` so they never write state onto a skill the curator can't manage."""
+    """Load, apply *mutator(record)* in place, save; returns the mutator result (None if nothing landed). Telemetry
+    is recorded for ANY skill; lifecycle mutators pass ``require_curation_eligible=True`` so state never lands on
+    a skill the curator can't manage."""
     if not skill_name:
         return None
     guard = (lambda: is_curation_eligible(skill_name)) if require_curation_eligible else None
@@ -475,19 +430,17 @@ def telemetry_provenance(skill_name: str, record: Optional[Dict[str, Any]] = Non
             if get_plugin_manager().find_plugin_skill(skill_name) is not None:
                 return "installed"
     created_by = record.get("created_by") if isinstance(record, dict) else None
-    if created_by in ("installed", "agent"):
-        return {"installed": "installed", "agent": "agent_created"}[created_by]
+    if label := {"installed": "installed", "agent": "agent_created"}.get(created_by):
+        return label
     if _find_external_skill_dir(skill_name) is not None:
         return "external"
-    if _find_skill_dir(skill_name) is not None or isinstance(record, dict):
-        return "local"
-    return "unknown"
+    return "local" if _find_skill_dir(skill_name) is not None or isinstance(record, dict) else "unknown"
 
 
-def _emit_skill_lifecycle(
-    skill_name: str, action: str, *, record: Optional[Dict[str, Any]] = None,
-    task_id: Optional[str] = None, session_id: Optional[str] = None, **facts: Any) -> None:
-    """Best-effort lifecycle hook after an authoritative state change; absent facts are sent as None."""
+def _emit_skill_lifecycle(skill_name: str, action: str, *, record: Optional[Dict[str, Any]] = None,
+                          task_id: Optional[str] = None, session_id: Optional[str] = None) -> None:
+    """Best-effort lifecycle hook after an authoritative state change; facts absent from *record* go as None."""
+    facts = record or {}
     try:
         from hermes_cli.lifecycle import has_hook, invoke_hook
         if has_hook("on_skill_lifecycle"):
@@ -504,7 +457,6 @@ def _mutate_and_emit(skill_name: str, action: str, mutator: Callable[[Dict[str,
     """``_mutate`` then emit *action* with the mutator's facts as the record — only if the write landed."""
     facts = _mutate(skill_name, mutator)
     if isinstance(facts, dict):
-        hook_kwargs.update({k: facts[k] for k in ("use_count", "reused", "reuse_after_patch") if k in facts})
         _emit_skill_lifecycle(skill_name, action, record=facts, **hook_kwargs)
 
 
@@ -516,17 +468,14 @@ def bump_view(skill_name: str) -> None:
 def bump_use(skill_name: str, *, task_id: Optional[str] = None, session_id: Optional[str] = None) -> None:
     """Skill actively used (loaded into the prompt path / referenced from an assistant turn)."""
     def _apply(rec: Dict[str, Any]) -> Dict[str, Any]:
-        previous_use_count = _non_negative_int(rec.get("use_count"))
-        patch_generation = _non_negative_int(rec.get("patch_generation"))
-        last_reused_generation = min(_non_negative_int(rec.get("last_reused_patch_generation")), patch_generation)
-        reused = previous_use_count > 0
-        reuse_after_patch = reused and patch_generation > last_reused_generation
-        rec.update(
-            use_count=previous_use_count + 1, last_used_at=_now_iso(), patch_generation=patch_generation,
-            last_reused_patch_generation=patch_generation if reuse_after_patch else last_reused_generation)
-        return {
-            "created_by": rec.get("created_by"), "use_count": rec["use_count"],
-            "reused": reused, "reuse_after_patch": reuse_after_patch}
+        uses = _non_negative_int(rec.get("use_count"))
+        gen = _non_negative_int(rec.get("patch_generation"))
+        last_reused = min(_non_negative_int(rec.get("last_reused_patch_generation")), gen)
+        reuse_after_patch = uses > 0 and gen > last_reused
+        rec.update(use_count=uses + 1, last_used_at=_now_iso(), patch_generation=gen,
+                   last_reused_patch_generation=gen if reuse_after_patch else last_reused)
+        return {"created_by": rec.get("created_by"), "use_count": uses + 1, "reused": uses > 0,
+                "reuse_after_patch": reuse_after_patch}
     _mutate_and_emit(skill_name, "loaded", _apply, task_id=task_id, session_id=session_id)
 
 
@@ -542,11 +491,11 @@ def bump_patch(
 
 def record_created(
     skill_name: str, *, agent_created: bool, task_id: Optional[str] = None, session_id: Optional[str] = None) -> None:
-    """Persist creation provenance and emit a create fact. A successful create is a new logical skill even
-    if stale sidecar state survived an earlier deletion, so the record is reset."""
+    """Persist creation provenance and emit a create fact; the record is reset (a create is a new logical skill
+    even if stale sidecar state survived an earlier deletion)."""
     def _apply(rec: Dict[str, Any]) -> Dict[str, Any]:
         rec.clear()
-        rec.update(_empty_record(), **({"created_by": "agent"} if agent_created else {}))
+        rec.update(_empty_record(), created_by="agent" if agent_created else None)
         return {"created_by": rec["created_by"]}
     _mutate_and_emit(skill_name, "created", _apply, task_id=task_id, session_id=session_id)
 
@@ -565,39 +514,36 @@ def mark_agent_created(skill_name: str) -> None:
 
 
 def set_state(skill_name: str, state: str) -> None:
-    """Set lifecycle state; no-op for an invalid state or a skill the curator can't manage. Emits
-    archived/stale/restored (active<-archived); active<-stale emits nothing."""
+    """Set lifecycle state (no-op for invalid state / unmanageable skill). Emits archived/stale/restored
+    (active<-archived); active<-stale emits nothing."""
     if state not in _VALID_STATES:
         logger.debug("set_state: invalid state %r for %s", state, skill_name)
         return
 
     def _apply(rec: Dict[str, Any]) -> Dict[str, Any]:
-        previous_state = rec.get("state")
-        facts = {"changed": previous_state != state, "created_by": rec.get("created_by")}
-        if facts["changed"]:
+        previous = rec.get("state")
+        if previous != state:
             rec["state"] = state
             if state != STATE_STALE:
                 rec["archived_at"] = _now_iso() if state == STATE_ARCHIVED else None
-            facts["previous_state"] = previous_state
-        return facts
+        return {"changed": previous != state, "created_by": rec.get("created_by"), "previous_state": previous}
     facts = _mutate(skill_name, _apply, require_curation_eligible=True)
-    if not isinstance(facts, dict) or not facts.get("changed"):
+    if not isinstance(facts, dict) or not facts["changed"]:
         return
-    restored = state == STATE_ACTIVE and facts.get("previous_state") == STATE_ARCHIVED
+    restored = state == STATE_ACTIVE and facts["previous_state"] == STATE_ARCHIVED
     action = "restored" if restored else {STATE_ARCHIVED: "archived", STATE_STALE: "stale"}.get(state)
     if action is not None:
         _emit_skill_lifecycle(skill_name, action, record=facts)
 
 
 def set_pinned(skill_name: str, pinned: bool) -> bool:
-    """Set/clear the pin flag; False when the write did not land (not curation-eligible) so callers can
-    report failure instead of a false success."""
+    """Set/clear the pin flag; False when the write did not land (not curation-eligible)."""
     return _set_field(skill_name, "pinned", bool(pinned))
 
 
 def set_sync(skill_name: str, sync: bool) -> None:
-    """Sync is OPT-IN (``sync: true`` on the record, read by ``tools.skills_sync_client.list_synced_skill_names``);
-    gated on curation eligibility so bundled/hub/external skills can't be marked."""
+    """Opt-in ``sync`` flag (read by ``skills_sync_client.list_synced_skill_names``); curation-gated so bundled/hub/
+    external skills can't be marked."""
     _set_field(skill_name, "sync", bool(sync))
 
 
@@ -613,8 +559,8 @@ def forget(skill_name: str) -> None:
 
 # --- Archive / restore ---
 def _relocate(src: Path, dest: Path, skill_name: str, action: str, **capture_kwargs: Any) -> Tuple[bool, str]:
-    """Move *src* to *dest* for *action* ("archive" | "restore") inside an audit-ledger entry, then apply the
-    suppression + state side effects. Ledger capture is best-effort; rename falls back to shutil.move across devices."""
+    """Move *src* to *dest* for *action* ("archive" | "restore") inside a best-effort audit-ledger entry, then apply
+    suppression + state side effects; rename falls back to shutil.move across devices."""
     try:
         from tools import skill_ledger as _ledger
         _ledger_before = _ledger.capture_before(src, **capture_kwargs)
@@ -628,13 +574,10 @@ def _relocate(src: Path, dest: Path, skill_name: str, action: str, **capture_kwa
             shutil.move(str(src), str(dest))
         except Exception as e:
             return False, f"failed to {action}: {e}"
-    if action == "archive":
-        if is_bundled(skill_name):  # pruning a built-in only sticks if the re-seeder is told to leave it alone
-            _toggle_suppressed_name(skill_name, add=True)
-        set_state(skill_name, STATE_ARCHIVED)
-    else:
-        _toggle_suppressed_name(skill_name, add=False)
-        set_state(skill_name, STATE_ACTIVE)
+    archiving = action == "archive"
+    if not archiving or is_bundled(skill_name):  # pruning a built-in only sticks if the re-seeder skips it
+        _toggle_suppressed_name(skill_name, add=archiving)
+    set_state(skill_name, STATE_ARCHIVED if archiving else STATE_ACTIVE)
     with suppress(Exception):
         if _ledger is not None:
             _ledger.record_mutation(action, skill_name, before=_ledger_before or [], after_root=dest)
@@ -665,16 +608,16 @@ def archive_skill(skill_name: str) -> Tuple[bool, str]:
         return False, f"failed to create archive dir: {e}"
     dest = archive_root / skill_dir.name
     if dest.exists():
-        dest = archive_root / f"{skill_dir.name}-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}"
+        dest = dest.with_name(f"{skill_dir.name}-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}")
     # complete_package: consolidation may have re-homed support files first, so a disk-only capture can come
     # back hollow; the fill from the newest curator backup keeps rollback restorable.
     return _relocate(skill_dir, dest, skill_name, "archive", complete_package=True, skill=skill_name)
 
 
 def restore_skill(skill_name: str) -> Tuple[bool, str]:
-    """Move an archived skill back to the flat top-level layout (category nesting is NOT reconstructed). Refuses
-    names now colliding with a hub skill, or a bundled built-in unless ``curator.prune_builtins`` is on (then
-    restoring is the documented way to lift a prune) — either would shadow the upstream copy."""
+    """Move an archived skill back to the flat top-level layout (nesting NOT reconstructed). Refuses names now
+    colliding with a hub skill, or a bundled built-in unless ``curator.prune_builtins`` is on (restoring is the
+    documented way to lift a prune) — either would shadow the upstream copy."""
     if is_hub_installed(skill_name):
         return False, f"skill '{skill_name}' is now hub-installed; restore would shadow the upstream version"
     if is_bundled(skill_name) and not _prune_builtins_enabled():
@@ -682,9 +625,8 @@ def restore_skill(skill_name: str) -> Tuple[bool, str]:
     archive_root = _archive_dir()
     if not archive_root.exists():
         return False, "no archive directory"
-    # Exact name first (recursive: older archive paths left nested layouts), then the timestamped-duplicate
-    # fallback. Only "<skill>-YYYYMMDDHHMMSS" (14 digits) counts — a bare startswith("<skill>-") would let
-    # restoring "git" pull an archived "git-helpers" out and rename it, destroying the sibling's only copy.
+    # Exact name first (recursive: older archives left nested layouts), then the timestamped duplicate. Only
+    # "<skill>-YYYYMMDDHHMMSS" counts — a bare startswith("<skill>-") would let restoring "git" steal "git-helpers".
     dirs = [p for p in archive_root.rglob("*") if p.is_dir()]
     prefix = f"{skill_name}-"
     candidates = [p for p in dirs if p.name == skill_name] or sorted(
@@ -703,8 +645,8 @@ def _match_skill_dir(skill_mds: Iterable[Path], skill_name: str) -> Optional[Pat
 
 
 def _find_skill_dir(skill_name: str) -> Optional[Path]:
-    """Skill dir by frontmatter ``name`` (flat or category-nested). The gated index iterator makes org mirrors
-    resolve ONLY for the active org (stale ``_org/<other>/`` trees never match)."""
+    """Skill dir by frontmatter ``name`` (flat or nested); the gated index iterator resolves org mirrors ONLY for
+    the active org."""
     base = _skills_dir()
     if not base.exists():
         return None
@@ -716,21 +658,19 @@ def _find_skill_dir(skill_name: str) -> Optional[Path]:
 def _find_external_skill_dir(skill_name: str) -> Optional[Path]:
     """Skill dir under configured external dirs by frontmatter name."""
     from agent.skill_utils import get_all_skills_dirs
-    for base in (b for b in get_all_skills_dirs()[1:] if b.exists()):
-        found = _match_skill_dir((p for p in base.rglob("SKILL.md") if not is_excluded_skill_path(p)), skill_name)
-        if found is not None:
-            return found
-    return None
+    return next((found for base in get_all_skills_dirs()[1:] if base.exists()
+                 if (found := _match_skill_dir((p for p in base.rglob("SKILL.md") if not is_excluded_skill_path(p)),
+                                               skill_name)) is not None), None)
 
 
 # --- Reporting — for the curator CLI / slash command ---
 def curated_report() -> List[Dict[str, Any]]:
-    """One backfilled row per curator-managed skill with ``provenance`` ('agent'|'bundled'|'hub') and ``_persisted``
-    (real record exists; fresh backfills get their inactivity clock seeded instead of counting as ancient)."""
+    """One backfilled row per curator-managed skill with ``provenance`` and ``_persisted`` (a real record exists;
+    fresh backfills get their inactivity clock seeded instead of counting as ancient)."""
     data = load_usage()
     names = set(list_agent_created_skill_names())
-    # A pinned-but-unmanaged skill must stay visible or its pin silently vanishes from `curator status`; the
-    # local-dir guard keeps stale records for deleted dirs from rendering as ghost rows (`curator unpin` cleans up).
+    # Pinned-but-unmanaged skills stay visible or their pin silently vanishes from `curator status`; the local-dir
+    # guard keeps stale records for deleted dirs from rendering as ghost rows.
     names.update(name for name, rec in data.items()
                  if rec.get("pinned") and is_curation_eligible(name) and _find_skill_dir(name) is not None)
     return [_report_row(n, data.get(n), _persisted=n in data, provenance=provenance(n)) for n in sorted(names)]
@@ -742,11 +682,10 @@ def provenance(skill_name: str) -> str:
 
 
 def usage_report() -> List[Dict[str, Any]]:
-    """Usage rows for EVERY skill on disk (built-ins and hub included), with ``provenance`` and ``_persisted`` —
-    unlike ``curated_report()``, which is scoped to curator-managed candidates."""
+    """Usage rows for EVERY skill on disk (built-ins and hub included); ``curated_report()`` is the managed subset."""
     base = _skills_dir()
     if not base.exists():
         return []
     data = load_usage()
-    names = sorted({name for name, _skill_md in _iter_skill_mds(base, local_only=False)})
-    return [_report_row(n, data.get(n), provenance=provenance(n), _persisted=n in data) for n in names]
+    return [_report_row(n, data.get(n), provenance=provenance(n), _persisted=n in data)
+            for n in sorted({name for name, _md in _iter_skill_mds(base, local_only=False)})]

tools/skillevaluator_scan.py

@@ -1,10 +1,10 @@
 #!/usr/bin/env python3
-"""Advisory NVIDIA SkillEvaluator Tier 1 scan for skill installs — runs alongside (never instead of)
-``tools/skills_guard.py``, the enforcement layer. Contract: warn, don't block (PII findings are shown, the
-install continues — the upstream PII scanner false-positives on ``git@github.com`` / ``op://``); prompt only
-for secrets-class criticals (``--force`` / non-interactive proceed with a loud warning); never break installs
-(missing/crashing/timed-out/unparseable scanner = no-op). Toggle: ``skills.tier1_advisory`` (default on).
-Binary: ``uv tool install --python 3.13 "skillevaluator @ git+https://github.com/NVIDIA/SkillEvaluator.git@v0.1.0"``."""
+"""Advisory NVIDIA SkillEvaluator Tier 1 scan for skill installs — alongside (never instead of) skills_guard,
+the enforcement layer. Contract: warn, don't block (the upstream PII scanner false-positives on
+``git@github.com`` / ``op://``); prompt only for secrets-class criticals (``--force`` / non-interactive proceed
+with a loud warning); a missing/crashing/timed-out/unparseable scanner is a no-op. Toggle
+``skills.tier1_advisory`` (default on). Binary: ``uv tool install --python 3.13
+"skillevaluator @ git+https://github.com/NVIDIA/SkillEvaluator.git@v0.1.0"``."""
 
 from __future__ import annotations
 
@@ -68,19 +68,13 @@ class Tier1Report:
         return [f for f in self.findings if f.is_secrets_class]
 
 
-def scanner_available() -> bool:
-    return shutil.which(SCANNER_BIN) is not None
-
-
 def tier1_advisory_enabled() -> bool:
     """``skills.tier1_advisory`` (default True; safe because the scan is a no-op without the binary)."""
     try:
         from hermes_cli.config import load_config
         skills_cfg = load_config().get("skills") or {}
         value = skills_cfg.get("tier1_advisory", True) if isinstance(skills_cfg, dict) else True
-        if isinstance(value, str):
-            return value.strip().lower() not in ("false", "0", "no", "off")
-        return bool(value)
+        return value.strip().lower() not in ("false", "0", "no", "off") if isinstance(value, str) else bool(value)
     except Exception:
         return True
 
@@ -97,42 +91,36 @@ def _parse_report(report: dict) -> Tier1Report:
             incomplete.append(validator)
         elif not res.get("passed", True):
             any_complete_failed = True
-        for f in res.get("findings", []) or []:
-            if not isinstance(f, dict):
-                continue
-            findings.append(Tier1Finding(
-                check=str(f.get("check_name", "")), validator=validator,
-                severity=str(f.get("severity", "info")).lower(), message=str(f.get("message", ""))[:200],
-                file=str(f.get("file_path", "")), line=int(f.get("line_number") or 0),
-                suggestion=str(f.get("suggestion", ""))[:200]))
+        findings.extend(Tier1Finding(
+            check=str(f.get("check_name", "")), validator=validator, severity=str(f.get("severity", "info")).lower(),
+            message=str(f.get("message", ""))[:200], file=str(f.get("file_path", "")),
+            line=int(f.get("line_number") or 0), suggestion=str(f.get("suggestion", ""))[:200])
+            for f in res.get("findings", []) or [] if isinstance(f, dict))
     return Tier1Report(available=True, passed=not any_complete_failed and not findings, findings=findings,
                        incomplete_checks=incomplete)
 
 
 def run_tier1_scan(skill_dir: Path, timeout: int = SCAN_TIMEOUT_SECONDS) -> Tier1Report:
     """Run SkillEvaluator Tier 1 over one skill dir; any failure returns ``available=False``, never raises."""
-    if not scanner_available():
+    if shutil.which(SCANNER_BIN) is None:
         return Tier1Report(available=False, error="scanner not on PATH")
+    unavailable = lambda why: Tier1Report(available=False, error=why)  # noqa: E731
     with tempfile.TemporaryDirectory(prefix="se-tier1-") as outdir:
         try:
-            subprocess.run(
-                [SCANNER_BIN, "validate", str(skill_dir), "--checks", TIER1_CHECKS, "--no-dedup",
-                 "-r", "json", "-o", outdir],
-                capture_output=True, text=True, timeout=timeout)
+            subprocess.run([SCANNER_BIN, "validate", str(skill_dir), "--checks", TIER1_CHECKS, "--no-dedup",
+                            "-r", "json", "-o", outdir], capture_output=True, text=True, timeout=timeout)
         except subprocess.TimeoutExpired:
-            return Tier1Report(available=False, error=f"scan timed out after {timeout}s")
+            return unavailable(f"scan timed out after {timeout}s")
         except OSError as exc:
-            return Tier1Report(available=False, error=f"scanner failed to launch: {exc}")
+            return unavailable(f"scanner failed to launch: {exc}")
         reports = sorted(Path(outdir).glob("skillevaluator-output-*.json"))
         if not reports:
-            return Tier1Report(available=False, error="scanner produced no JSON report")
+            return unavailable("scanner produced no JSON report")
         try:
             parsed = json.loads(reports[-1].read_text(encoding="utf-8"))
         except (json.JSONDecodeError, OSError) as exc:
-            return Tier1Report(available=False, error=f"unparseable report: {exc}")
-        if not isinstance(parsed, dict):
-            return Tier1Report(available=False, error="unexpected report shape")
-        return _parse_report(parsed)
+            return unavailable(f"unparseable report: {exc}")
+        return _parse_report(parsed) if isinstance(parsed, dict) else unavailable("unexpected report shape")
 
 
 def format_tier1_report(report: Tier1Report, limit: int = 10) -> str:
@@ -144,13 +132,11 @@ def format_tier1_report(report: Tier1Report, limit: int = 10) -> str:
         lines.append("SkillEvaluator Tier 1: no findings from completed checks." if report.incomplete_checks
                      else "SkillEvaluator Tier 1: no findings.")
     else:
-        lines.append(
-            f"SkillEvaluator Tier 1 (advisory): "
-            f"{len(report.findings)} finding(s) — informational, verify before relying on this skill.")
+        lines.append(f"SkillEvaluator Tier 1 (advisory): {len(report.findings)} finding(s) — informational, "
+                     "verify before relying on this skill.")
         shown = report.secrets_findings + report.advisory_findings
-        for f in shown[:limit]:
-            tag = "SECRETS" if f.is_secrets_class else f.severity.upper()
-            lines.append(f"  [{tag}] {f.location()} — {f.message}")
+        lines.extend(f"  [{'SECRETS' if f.is_secrets_class else f.severity.upper()}] {f.location()} — {f.message}"
+                     for f in shown[:limit])
         if len(shown) > limit:
             lines.append(f"  … and {len(shown) - limit} more")
     if report.incomplete_checks:

tools/skills_ast_audit.py

@@ -1,7 +1,6 @@
 """AST-level deep audit for skill Python files — opt-in diagnostic (``hermes skills audit --deep``), not a
-security gate (SECURITY.md §2.4). Flags dynamic import / dynamic attribute access patterns for human review;
-every pattern has legitimate uses, so findings are hints, not verdicts.
-"""
+security gate (SECURITY.md §2.4). Flags dynamic import / attribute access for human review; every pattern has
+legitimate uses, so findings are hints, not verdicts."""
 
 from __future__ import annotations
 
@@ -30,9 +29,7 @@ def _scan_source(content: str, rel_path: str) -> List[Finding]:
     except (SyntaxError, ValueError, RecursionError):
         return []
     findings: List[Finding] = []
-
-    def hit(node, pid: str, desc: str) -> None:
-        findings.append((rel_path, node.lineno, pid, desc))
+    hit = lambda node, pid, desc: findings.append((rel_path, node.lineno, pid, desc))  # noqa: E731
 
     class V(ast.NodeVisitor):
         def visit_Call(self, node):
@@ -58,9 +55,8 @@ def _scan_source(content: str, rel_path: str) -> List[Finding]:
             self.generic_visit(node)
 
         def visit_ImportFrom(self, node):
-            m = node.module or ""
-            if _is_importlib(m):
-                hit(node, "importlib_import", f"from {m} import ... — enables dynamic module loading")
+            if _is_importlib(node.module or ""):
+                hit(node, "importlib_import", f"from {node.module} import ... — enables dynamic module loading")
             self.generic_visit(node)
 
     try:
@@ -83,16 +79,8 @@ def ast_scan_path(path: Path) -> List[Finding]:
         return _scan_file(path, path.name) if path.suffix.lower() == ".py" else []
     if not path.is_dir():
         return []
-    out: List[Finding] = []
-    for py in sorted(path.rglob("*.py")):
-        if set(py.parent.parts) & _IGNORED_DIRS:
-            continue
-        try:
-            rel = py.relative_to(path).as_posix()
-        except ValueError:
-            rel = py.name
-        out.extend(_scan_file(py, rel))
-    return out
+    return [f for py in sorted(path.rglob("*.py")) if not set(py.parent.parts) & _IGNORED_DIRS
+            for f in _scan_file(py, py.relative_to(path).as_posix())]
 
 
 def format_ast_report(findings: List[Finding], skill_name: str = "") -> str:
@@ -107,5 +95,4 @@ def format_ast_report(findings: List[Finding], skill_name: str = "") -> str:
             current = f
             lines.append(f"  {f}")
         lines.append(f"    L{line}  {pid}  — {desc}")
-    lines += ["", "  Note: diagnostic hints for human review, not security verdicts."]
-    return "\n".join(lines)
+    return "\n".join(lines + ["", "  Note: diagnostic hints for human review, not security verdicts."])

tools/skills_guard.py

@@ -1,12 +1,11 @@
 #!/usr/bin/env python3
 """Skills Guard — regex static scan of externally-sourced skills plus a trust-aware install policy.
 
-Trust: builtin (ships with Hermes, never scanned), trusted (openai/anthropics/... repos: caution allowed),
-community (any findings block unless --force). ``scan_skill(path, source)`` -> ``should_allow_install`` ->
-``format_scan_report``. Known limitation: language write APIs (open(..., 'w'), Path.write_text, shutil.copy*,
-fs.writeFileSync) aimed at agent-config files surface only the low *_ref finding — static regexes cannot tie
-the call to a dynamically built destination; any future coverage belongs as a fourth "mechanical" tier next
-to agent_config_mod_shell."""
+Trust: builtin (never scanned), trusted (openai/anthropics/... repos: caution allowed), community (any
+findings block unless --force). ``scan_skill`` -> ``should_allow_install`` -> ``format_scan_report``.
+Known gap: language write APIs (open(..., 'w'), Path.write_text, shutil.copy*, fs.writeFileSync) aimed at
+agent-config files surface only the low *_ref finding — static regexes cannot tie the call to a dynamic
+destination; future coverage belongs as a fourth "mechanical" tier next to agent_config_mod_shell."""
 
 import re
 import fnmatch
@@ -40,7 +39,7 @@ VERDICT_INDEX = {"safe": 0, "caution": 1, "dangerous": 2}
 class Finding:
     pattern_id: str
     severity: str       # "critical" | "high" | "medium" | "low"
-    category: str       # "exfiltration" | "injection" | "destructive" | "persistence" | "network" | "obfuscation"
+    category: str       # "exfiltration" | "injection" | "destructive" | "persistence" | "network" | ...
     file: str
     line: int
     match: str
@@ -390,9 +389,9 @@ def _unicode_char_name(char: str) -> str:
 
 
 def _compute_docstring_lines(lines: list) -> set:
-    """1-indexed line numbers inside (or on the boundary of) triple-quoted strings: opening, interior, closing and
-    self-contained one-line docstrings all count, so ``os.environ`` in prose is not scored. Heuristic (ignores a
-    triple quote inside a string literal) but covers the common false-positive shapes."""
+    """1-indexed lines inside or on the boundary of triple-quoted strings (opening, interior, closing, and
+    one-line docstrings), so ``os.environ`` in prose is not scored. Heuristic: a triple quote inside a string
+    literal is miscounted, but the common false-positive shapes are covered."""
     doc_lines: set = set()
     in_docstring = False
     for i, line in enumerate(lines):
@@ -405,17 +404,16 @@ def _compute_docstring_lines(lines: list) -> set:
 
 
 def scan_file(file_path: Path, rel_path: str = "") -> List[Finding]:
-    """Scan one file for threat patterns and invisible unicode. *rel_path* is the display path (defaults
-    to the file name). Regex findings dedupe per pattern per line; invisible chars yield one per line."""
+    """Threat-pattern + invisible-unicode scan of one file; *rel_path* is the display path (default: file
+    name). Regex findings dedupe per pattern per line; invisible chars yield one per line."""
     rel_path = rel_path or file_path.name
     if file_path.suffix.lower() not in SCANNABLE_EXTENSIONS and file_path.name != "SKILL.md":
         return []
     try:
-        content = file_path.read_text(encoding='utf-8')
+        lines = file_path.read_text(encoding='utf-8').split('\n')
     except (UnicodeDecodeError, OSError):
         return []
     findings = []
-    lines = content.split('\n')
     docstring_lines = _compute_docstring_lines(lines)  # so code patterns don't fire on prose
     for pattern, pid, severity, category, description in _COMPILED_THREAT_PATTERNS:
         for i, line in enumerate(lines, start=1):
@@ -424,91 +422,79 @@ def scan_file(file_path: Path, rel_path: str = "") -> List[Finding]:
                 findings.append(Finding(pid, severity, category, rel_path, i,
                                         text if len(text) <= 120 else text[:117] + "...", description))
     for i, line in enumerate(lines, start=1):
-        char = next((c for c in INVISIBLE_CHARS if c in line), None)
-        if char is not None:
-            char_name = _unicode_char_name(char)
-            findings.append(Finding(
-                "invisible_unicode", "high", "injection", rel_path, i, f"U+{ord(char):04X} ({char_name})",
-                f"invisible unicode character {char_name} (possible text hiding/injection)"))
+        if (char := next((c for c in INVISIBLE_CHARS if c in line), None)) is not None:
+            name = _unicode_char_name(char)
+            findings.append(Finding("invisible_unicode", "high", "injection", rel_path, i,
+                                    f"U+{ord(char):04X} ({name})",
+                                    f"invisible unicode character {name} (possible text hiding/injection)"))
     return findings
 
 
 def scan_skill(skill_path: Path, source: str = "community") -> ScanResult:
-    """Structural checks + pattern scan of every text file in a skill dir (or a single file). A `.skillignore` /
-    `.clawhubignore` (gitignore-style) excludes dev/docs artifacts from BOTH passes; the ignore file itself is
+    """Structural checks + pattern scan of every text file in a skill dir (or a single file). A gitignore-style
+    `.skillignore` / `.clawhubignore` excludes dev/docs artifacts from BOTH passes; the ignore file itself is
     always excluded and `SKILL.md` can never be un-ignored. *source* (e.g. "openai/skills") sets the trust level."""
-    skill_name = skill_path.name
-    trust_level = _resolve_trust_level(source)
-    all_findings: List[Finding] = []
+    name, trust = skill_path.name, _resolve_trust_level(source)
+    findings: List[Finding] = []
     if skill_path.is_dir():
         ignore = _load_skill_ignore(skill_path)
-        all_findings.extend(_check_structure(skill_path, ignore=ignore))
+        findings.extend(_check_structure(skill_path, ignore=ignore))
         for f in skill_path.rglob("*"):
             rel = str(f.relative_to(skill_path))
             if f.is_file() and not ignore(rel):
-                all_findings.extend(scan_file(f, rel))
+                findings.extend(scan_file(f, rel))
     elif skill_path.is_file():
-        all_findings.extend(scan_file(skill_path, skill_path.name))
-    verdict = _determine_verdict(all_findings)
-    return ScanResult(
-        skill_name=skill_name, source=source, trust_level=trust_level, verdict=verdict, findings=all_findings,
-        scanned_at=datetime.now(timezone.utc).isoformat(),
-        summary=_build_summary(skill_name, source, trust_level, verdict, all_findings),
-    )
+        findings.extend(scan_file(skill_path, skill_path.name))
+    verdict = _determine_verdict(findings)
+    return ScanResult(name, source, trust, verdict, findings, datetime.now(timezone.utc).isoformat(),
+                      _build_summary(name, source, trust, verdict, findings))
 
 
 def _content_digest(skill_path: Path) -> str:
-    """Canonical SHA-256 over (POSIX relative path, file bytes), ORDERED by the rel-path STRING: sorting Paths is
-    case-insensitive on Windows and diverged from ``tools.skills_hub.bundle_content_hash`` (plain-string sort), so
-    every installed skill reported ``update_available`` forever. String order keeps both sides byte-symmetric."""
+    """Canonical SHA-256 over (POSIX relative path, file bytes) ORDERED by the rel-path STRING — Path sorting is
+    case-insensitive on Windows and diverged from ``skills_hub.bundle_content_hash`` (every installed skill then
+    reported ``update_available`` forever). String order keeps both sides byte-symmetric."""
     h = hashlib.sha256()
     if not skill_path.is_dir():
         h.update(skill_path.read_bytes())
         return h.hexdigest()
-    entries = sorted((p.relative_to(skill_path).as_posix(), p) for p in skill_path.rglob("*") if p.is_file())
-    for rel, file_path in entries:
+    for rel, p in sorted((p.relative_to(skill_path).as_posix(), p) for p in skill_path.rglob("*") if p.is_file()):
         h.update(rel.encode("utf-8") + b"\x00")
-        h.update(file_path.read_bytes())
+        h.update(p.read_bytes())
     return h.hexdigest()
 
 
-def full_content_hash(skill_path: Path) -> str:
-    """Full canonical digest used to bind scanner attestations."""
-    return f"sha256:{_content_digest(skill_path)}"
-
-
 def content_hash(skill_path: Path) -> str:
-    """Short integrity hash. Paths are mixed in so swapping two files' contents changes it. MUST stay
-    symmetric with ``tools.skills_hub.bundle_content_hash`` (disk vs in-memory bundle) — change both at once."""
+    """Short integrity hash (paths mixed in, so swapping two files' contents changes it). MUST stay symmetric
+    with ``tools.skills_hub.bundle_content_hash`` — change both at once."""
     return f"sha256:{_content_digest(skill_path)[:16]}"
 
 
 def scan_skill_cached(
     skill_path: Path, source: str = "community", *, source_url: str = "", cache_dir: Path | None = None,
 ) -> Tuple[ScanResult, dict]:
-    """Return a scan plus attestation, caching only exact current content."""
-    bundle_hash = full_content_hash(skill_path)
+    """Scan plus attestation dict; the cache (keyed by content digest + source identity) only serves exact
+    current content under the current scanner version."""
+    digest = _content_digest(skill_path)
     cache_root = cache_dir or skill_path.parent / ".scan-cache"
     source_identity = hashlib.sha256(f"{source}\0{source_url}".encode("utf-8")).hexdigest()[:16]
-    cache_file = cache_root / f"{bundle_hash.split(':', 1)[1]}-{source_identity}.json"
-    try:
+    cache_file = cache_root / f"{digest}-{source_identity}.json"
+    expected = {"bundle_hash": f"sha256:{digest}", "scanner_version": SCANNER_VERSION, "source": source,
+                "source_url": source_url}
+    cached = None
+    with suppress(OSError, json.JSONDecodeError):
         cached = json.loads(cache_file.read_text(encoding="utf-8"))
-    except (OSError, json.JSONDecodeError):
-        cached = None
-    expected = {"bundle_hash": bundle_hash, "scanner_version": SCANNER_VERSION, "source": source, "source_url": source_url}
     if isinstance(cached, dict) and all(cached.get(k) == v for k, v in expected.items()):
-        result = ScanResult(
-            skill_name=skill_path.name, source=source, trust_level=cached["trust_level"], verdict=cached["verdict"],
-            findings=[Finding(**item) for item in cached.get("findings", [])],
-            scanned_at=cached["scanned_at"], summary=cached.get("summary", ""))
+        result = ScanResult(skill_path.name, source, cached["trust_level"], cached["verdict"],
+                            [Finding(**item) for item in cached.get("findings", [])], cached["scanned_at"],
+                            cached.get("summary", ""))
         provenance = {**cached, "fresh": False}
     else:
         result = scan_skill(skill_path, source=source)
         findings = [asdict(item) for item in result.findings]
-        provenance = {
-            **expected, "verdict": result.verdict, "trust_level": result.trust_level, "findings": findings,
-            "rules": sorted({item["pattern_id"] for item in findings}),
-            "scanned_at": result.scanned_at, "summary": result.summary, "fresh": True}
+        provenance = {**expected, "verdict": result.verdict, "trust_level": result.trust_level, "findings": findings,
+                      "rules": sorted({item["pattern_id"] for item in findings}), "scanned_at": result.scanned_at,
+                      "summary": result.summary, "fresh": True}
         with suppress(OSError):
             cache_root.mkdir(parents=True, exist_ok=True)
             cache_file.write_text(json.dumps(provenance, indent=2) + "\n", encoding="utf-8")
@@ -517,8 +503,8 @@ def scan_skill_cached(
 
 
 def should_allow_install(result: ScanResult, force: bool = False) -> Tuple[bool, str]:
-    """Decide install from verdict + trust; ``(allowed, reason)``. *force* overrides every block except a
-    dangerous verdict on community/trusted sources. ``allowed`` is None when policy says "ask"."""
+    """``(allowed, reason)`` from verdict + trust; *force* overrides every block except a dangerous verdict on
+    community/trusted sources. ``allowed`` is None when policy says "ask"."""
     policy = INSTALL_POLICY.get(result.trust_level, INSTALL_POLICY["community"])
     decision = policy[VERDICT_INDEX.get(result.verdict, 2)]
     n = len(result.findings)
@@ -540,10 +526,10 @@ def format_scan_report(result: ScanResult) -> str:
     """Compact multi-line report for CLI/chat display; findings sorted critical → low."""
     lines = [f"Scan: {result.skill_name} ({result.source}/{result.trust_level})  Verdict: {result.verdict.upper()}"]
     if result.findings:
-        severity_order = {"critical": 0, "high": 1, "medium": 2, "low": 3}
-        for f in sorted(result.findings, key=lambda f: severity_order.get(f.severity, 4)):
-            loc = f"{f.file}:{f.line}".ljust(30)
-            lines.append(f"  {f.severity.upper().ljust(8)} {f.category.ljust(14)} {loc} \"{f.match[:60]}\"")
+        order = {"critical": 0, "high": 1, "medium": 2, "low": 3}
+        for f in sorted(result.findings, key=lambda f: order.get(f.severity, 4)):
+            lines.append(f"  {f.severity.upper().ljust(8)} {f.category.ljust(14)} "
+                         f"{f'{f.file}:{f.line}'.ljust(30)} \"{f.match[:60]}\"")
         lines.append("")
     allowed, reason = should_allow_install(result)
     status = "ALLOWED" if allowed is True else "NEEDS CONFIRMATION" if allowed is None else "BLOCKED"
@@ -552,18 +538,16 @@ def format_scan_report(result: ScanResult) -> str:
 
 
 def _check_structure(skill_dir: Path, ignore=None) -> List[Finding]:
-    """Structural anomalies: file count, total size, binary/executable files, symlinks escaping the skill
-    dir, oversized files. *ignore(rel_path) -> bool* excludes paths from every count and finding."""
-    if ignore is None:
-        ignore = lambda _rel: False  # noqa: E731
+    """Structural anomalies (counts, sizes, binaries, stray executables, escaping symlinks); *ignore(rel) -> bool*
+    excludes paths from every count and finding."""
     findings = []
 
-    def add(pid: str, severity: str, category: str, rel: str, match: str, description: str) -> None:
-        findings.append(Finding(pid, severity, category, rel, 0, match, description))
+    def add(pid, sev, cat, rel, match, desc):
+        findings.append(Finding(pid, sev, cat, rel, 0, match, desc))
     file_count = total_size = 0
     for f in skill_dir.rglob("*"):
         rel = str(f.relative_to(skill_dir))
-        if not (f.is_file() or f.is_symlink()) or ignore(rel):
+        if not (f.is_file() or f.is_symlink()) or (ignore is not None and ignore(rel)):
             continue
         file_count += 1
         if f.is_symlink():
@@ -612,37 +596,31 @@ def _load_skill_ignore(skill_dir: Path):
     ``SKILL.md`` never."""
     patterns: List[str] = []
     for ig in (skill_dir / name for name in _SKILL_IGNORE_FILENAMES):
-        try:
+        with suppress(UnicodeDecodeError, OSError):
             if ig.is_file():
                 lines = (raw.strip() for raw in ig.read_text(encoding="utf-8").splitlines())
                 patterns.extend(line for line in lines if line and not line.startswith("#"))
-        except (UnicodeDecodeError, OSError):
-            continue
 
     def ignore(rel: str) -> bool:
         rel_posix = Path(rel).as_posix()
-        base = rel_posix.split("/")[-1]
+        segs = rel_posix.split("/")
+        base = segs[-1]
         if base in _NEVER_IGNORABLE:
             return False
         if base in _ALWAYS_IGNORED_NAMES:
             return True
         for pat in patterns:
             anchored = pat.startswith("/")
-            p = pat.lstrip("/")
-            is_dir = p.endswith("/")
-            p = p.rstrip("/")
+            p = pat.strip("/")
             if not p:
                 continue
             below = rel_posix.startswith(p + "/")
-            if is_dir:  # the dir itself or anything under it; unanchored also as an inner path component
+            if pat.endswith("/"):  # the dir itself or anything under it; unanchored also as an inner path component
                 if rel_posix == p or below or (not anchored and ("/" + p + "/") in ("/" + rel_posix + "/")):
                     return True
-                continue
-            if fnmatch.fnmatch(rel_posix, p):
-                return True
             # Unanchored: also the basename, any path segment, or a prefix dir (`docs` ignores docs/plans/x.md).
-            if not anchored and (fnmatch.fnmatch(base, p) or below
-                                 or ("/" not in p and any(fnmatch.fnmatch(seg, p) for seg in rel_posix.split("/")))):
+            elif fnmatch.fnmatch(rel_posix, p) or (not anchored and (fnmatch.fnmatch(base, p) or below or (
+                    "/" not in p and any(fnmatch.fnmatch(seg, p) for seg in segs)))):
                 return True
         return False
 
@@ -653,18 +631,14 @@ _SOURCE_PREFIX_ALIASES = ("skills-sh/", "skills.sh/", "skils-sh/", "skils.sh/")
 
 
 def _resolve_trust_level(source: str) -> str:
-    """Map a source identifier to a trust level. "official" is provenance, not a user-controlled GitHub
-    id like "official/<repo>". Trusted repos match exactly or as a skill path inside the repo — never a
-    sibling repo sharing the prefix."""
-    prefix = next((p for p in _SOURCE_PREFIX_ALIASES if source.startswith(p)), "")
-    normalized_source = source[len(prefix):]
-    if normalized_source == "agent-created":
+    """Source id -> trust level. "official" is provenance, not a user-controlled GitHub id like "official/<repo>";
+    trusted repos match exactly or as a skill path inside the repo — never a sibling sharing the prefix."""
+    src = source[len(next((p for p in _SOURCE_PREFIX_ALIASES if source.startswith(p)), "")):]
+    if src == "agent-created":
         return "agent-created"
-    if normalized_source == "official":
+    if src == "official":
         return "builtin"
-    if any(normalized_source == t or normalized_source.startswith(f"{t}/") for t in TRUSTED_REPOS):
-        return "trusted"
-    return "community"
+    return "trusted" if any(src == t or src.startswith(f"{t}/") for t in TRUSTED_REPOS) else "community"
 
 
 def _determine_verdict(findings: List[Finding]) -> str: