refactor(tools): skills group pass 2b — skill_usage state/relocate/archive collapse, _backfilled merge, dedup ints; skills_guard ignore constants inlined, verdict/report joins, single stat; ast_audit/skillevaluator flattening

tools/skill_usage.py

@@ -120,6 +120,10 @@ def _int_or_zero(value: Any) -> int:
         return 0
 
 
+def _non_negative_int(value: Any) -> int:
+    return 0 if isinstance(value, bool) else max(0, _int_or_zero(value))
+
+
 def activity_count(record: Dict[str, Any]) -> int:
     """Total observed use+view+patch events."""
     return sum(_int_or_zero(record.get(key)) for key in ("use_count", "view_count", "patch_count"))
@@ -136,12 +140,10 @@ def _read_hub_installed_names() -> Set[str]:
     """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"))
+        data = json.loads(lock_path.read_text(encoding="utf-8", errors="replace")) if lock_path.exists() else {}
     except (OSError, json.JSONDecodeError) as e:
         logger.debug("Failed to read hub lock file: %s", e)
         return set()
@@ -202,9 +204,8 @@ def _scan_local_skills(keep: Callable[[str, Path, Set[str], Dict[str, Any]], boo
     if not base.exists():
         return []
     hub, bundled, usage = _read_hub_installed_names(), _read_bundled_manifest_names(), load_usage()
-    return sorted({
-        name for name, skill_md in _iter_skill_mds(base, local_only=True)
-        if name not in hub and not is_protected_builtin(name) and keep(name, skill_md, bundled, usage)})
+    return sorted({name for name, skill_md in _iter_skill_mds(base, local_only=True)
+                   if name not in hub and not is_protected_builtin(name) and keep(name, skill_md, bundled, usage)})
 
 
 def list_agent_created_skill_names() -> List[str]:
@@ -230,8 +231,7 @@ def _read_skill_name(skill_md: Path, fallback: str) -> str:
     if "---" not in lines:
         return fallback
     block = lines[lines.index("---") + 1:]  # frontmatter runs to the closing --- or (truncated) end of text
-    if "---" in block:
-        block = block[:block.index("---")]
+    block = block[:block.index("---")] if "---" in block else block
     values = (line.split(":", 1)[1].strip().strip("\"'") for line in block if line.startswith("name:"))
     return next((v for v in values if v), fallback)
 
@@ -306,9 +306,9 @@ def adopt_skill(skill_name: str) -> Tuple[bool, str]:
     if is_bundled(skill_name):  # governed by prune_builtins; stamping created_by=agent would change nothing
         return False, f"'{skill_name}' is a bundled built-in — it is governed by curator.prune_builtins, not by adoption"
     skill_dir = _find_skill_dir(skill_name)
-    if skill_dir is None and _find_external_skill_dir(skill_name) is not None:
-        return False, f"'{skill_name}' lives in skills.external_dirs and is read-only to the curator"
     if skill_dir is None:
+        if _find_external_skill_dir(skill_name) is not None:
+            return False, f"'{skill_name}' lives in skills.external_dirs and is read-only to the curator"
         return False, f"skill '{skill_name}' not found"
     if is_external_skill_path(skill_dir):
         return False, _external_read_only_message(skill_name)
@@ -331,9 +331,7 @@ def _backfilled(rec: Any) -> Dict[str, Any]:
     """*rec* with every missing default key filled in (a fresh record when not a dict)."""
     if not isinstance(rec, dict):
         return _empty_record()
-    for k, v in _empty_record().items():
-        rec.setdefault(k, v)
-    return rec
+    return {**rec, **{k: v for k, v in _empty_record().items() if k not in rec}}  # rec's key order first
 
 
 def _report_row(name: str, raw: Any, **extra: Any) -> Dict[str, Any]:
@@ -369,9 +367,8 @@ def get_record(skill_name: str) -> Dict[str, Any]:
     return _backfilled(load_usage().get(skill_name))
 
 
-def _locked_update(
-    skill_name: str, op: Callable[[Dict[str, Dict[str, Any]]], Tuple[Any, bool]], fail_log: str,
-    guard: Optional[Callable[[], bool]] = None) -> Any:
+def _locked_update(skill_name: str, op: Callable[[Dict[str, Dict[str, Any]]], Tuple[Any, bool]], fail_log: str,
+                   guard: Optional[Callable[[], bool]] = None) -> Any:
     """Run *op(data) -> (result, dirty)* under the file lock, saving only when dirty; *guard* runs before locking.
     None when the guard failed, the save did not land, or anything raised (DEBUG-logged via *fail_log*)."""
     try:
@@ -401,9 +398,9 @@ def _mutate(skill_name: str, mutator, *, require_curation_eligible: bool = False
     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
     return _locked_update(skill_name, lambda data: (mutator(data.setdefault(skill_name, _empty_record())), True),
-                          "skill_usage._mutate(%s) failed: %s", guard)
+                          "skill_usage._mutate(%s) failed: %s",
+                          (lambda: is_curation_eligible(skill_name)) if require_curation_eligible else None)
 
 
 def _set_field(skill_name: str, key: str, value: Any) -> bool:
@@ -411,10 +408,6 @@ def _set_field(skill_name: str, key: str, value: Any) -> bool:
     return bool(_mutate(skill_name, lambda rec: rec.update({key: value}) or True, require_curation_eligible=True))
 
 
-def _non_negative_int(value: Any) -> int:
-    return 0 if isinstance(value, bool) else max(0, _int_or_zero(value))
-
-
 def _bump(rec: Dict[str, Any], count_key: str, ts_key: str) -> None:
     rec[count_key] = _non_negative_int(rec.get(count_key)) + 1
     rec[ts_key] = _now_iso()
@@ -479,18 +472,19 @@ def bump_use(skill_name: str, *, task_id: Optional[str] = None, session_id: Opti
     _mutate_and_emit(skill_name, "loaded", _apply, task_id=task_id, session_id=session_id)
 
 
-def bump_patch(
-    skill_name: str, *, action: str = "patch", task_id: Optional[str] = None, session_id: Optional[str] = None) -> None:
+def bump_patch(skill_name: str, *, action: str = "patch", task_id: Optional[str] = None,
+               session_id: Optional[str] = None) -> None:
     """Called from skill_manage (patch/edit)."""
     def _apply(rec: Dict[str, Any]) -> Dict[str, Any]:
         _bump(rec, "patch_count", "last_patched_at")
         rec["patch_generation"] = _non_negative_int(rec.get("patch_generation")) + 1
         return {"created_by": rec.get("created_by")}
-    _mutate_and_emit(skill_name, "patched" if action == "patch" else "edited", _apply, task_id=task_id, session_id=session_id)
+    _mutate_and_emit(skill_name, "patched" if action == "patch" else "edited", _apply, task_id=task_id,
+                     session_id=session_id)
 
 
-def record_created(
-    skill_name: str, *, agent_created: bool, task_id: Optional[str] = None, session_id: Optional[str] = None) -> None:
+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; 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]:
@@ -528,12 +522,11 @@ def set_state(skill_name: str, state: str) -> None:
                 rec["archived_at"] = _now_iso() if state == STATE_ARCHIVED else None
         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["changed"]:
-        return
-    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)
+    if isinstance(facts, dict) and facts["changed"]:
+        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:
@@ -600,13 +593,11 @@ def archive_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)
-
-    archive_root = _archive_dir()
+    dest = _archive_dir() / skill_dir.name
     try:
-        archive_root.mkdir(parents=True, exist_ok=True)
+        dest.parent.mkdir(parents=True, exist_ok=True)
     except OSError as e:
         return False, f"failed to create archive dir: {e}"
-    dest = archive_root / skill_dir.name
     if dest.exists():
         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

tools/skillevaluator_scan.py

@@ -84,19 +84,19 @@ def _parse_report(report: dict) -> Tier1Report:
     validators are kept (partial evidence is evidence) but excluded from the pass/fail signal."""
     findings: List[Tier1Finding] = []
     incomplete: List[str] = []
-    any_complete_failed = False
+    failed = False
     for res in report.get("results", []) or []:
         validator = str(res.get("validator", "unknown"))
         if str(res.get("status", "")).lower() == "incomplete":
             incomplete.append(validator)
-        elif not res.get("passed", True):
-            any_complete_failed = True
+        else:
+            failed = failed or not res.get("passed", True)
         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,
+    return Tier1Report(available=True, passed=not failed and not findings, findings=findings,
                        incomplete_checks=incomplete)
 
 

tools/skills_ast_audit.py

@@ -77,10 +77,8 @@ def ast_scan_path(path: Path) -> List[Finding]:
     """Scan one .py file or every .py under a directory; [] for non-Python/missing paths."""
     if path.is_file():
         return _scan_file(path, path.name) if path.suffix.lower() == ".py" else []
-    if not path.is_dir():
-        return []
     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())]
+            for f in _scan_file(py, py.relative_to(path).as_posix())] if path.is_dir() else []
 
 
 def format_ast_report(findings: List[Finding], skill_name: str = "") -> str:
@@ -88,8 +86,7 @@ def format_ast_report(findings: List[Finding], skill_name: str = "") -> str:
     header = f"AST deep scan: {skill_name}" if skill_name else "AST deep scan"
     if not findings:
         return f"{header}\n  No dynamic import/access patterns detected."
-    lines = [header, f"  {len(findings)} finding(s):"]
-    current = None
+    lines, current = [header, f"  {len(findings)} finding(s):"], None
     for f, line, pid, desc in sorted(findings):
         if f != current:
             current = f

tools/skills_guard.py

@@ -67,8 +67,7 @@ MODIFY_VERB_RE = (
     r'|\bmodif(?:y|ies|ied|ying|ication)s?\b|\bupdat(?:e|es|ed|ing)\b'
     r'|\bappend(?:s|ed|ing)?\b|\bprepend(?:s|ed|ing)?\b'
     r'|\binject(?:s|ed|ing)?\b|\boverwrit(?:e|es|ing)\b|\boverwritten\b'
-    r'|\breplac(?:e|es|ed|ing)\b|\balter(?:s|ed|ing)?\b|\badd(?:s|ed|ing)\b)'
-)
+    r'|\breplac(?:e|es|ed|ing)\b|\balter(?:s|ed|ing)?\b|\badd(?:s|ed|ing)\b)')
 
 _AGENT_CONFIG_FILES = r'(?:AGENTS\.md|CLAUDE\.md|\.cursorrules|\.clinerules)'
 _HERMES_CONFIG_FILES = r'\.hermes/(?:config\.yaml|SOUL\.md)'
@@ -78,15 +77,14 @@ _OTHER_AGENT_CONFIG_FILES = r'\.(?:claude/settings|codex/config)[\w.]*'
 
 def _shell_write_re(file_alt: str) -> str:
     """Mechanical shell write into *file_alt*: ``>``/``>>``, ``sed -i``, ``tee`` (target as immediate argument, so
-    ``| tee output | AGENTS.md |`` cells miss), ``cp``/``mv`` with the file as destination (a source arg is required,
-    so ``cp AGENTS.md backup/`` misses; ``AGENTS.md.bak`` is not the file). A single ``>`` needs a preceding
-    word/quote/paren char so blockquotes (``> text``) and arrows (``-> file``) miss."""
+    ``| tee output | AGENTS.md |`` cells miss), ``cp``/``mv`` with the file as destination (source arg required, so
+    ``cp AGENTS.md backup/`` misses; ``AGENTS.md.bak`` is not the file). A single ``>`` needs a preceding word/quote/
+    paren char so blockquotes (``> text``) and arrows (``-> file``) miss."""
     return (
         rf'(?:>>|[\w"\'`)\]]\s*>)\s*[~\w./-]*{file_alt}(?!\.?\w)'
         rf'|\bsed\b[^\n]*\s(?:-[A-Za-z]*i[A-Za-z]*|--in-place)\b[^\n]*{file_alt}(?!\.?\w)'
         rf'|\btee\s+(?:-a\s+)?[~\w./"\'-]*{file_alt}(?!\.?\w)'
-        rf'|\b(?:cp|mv)\s+[^\s|;&]+\s+[^\n|;&]{{0,40}}?{file_alt}(?!\.?\w)'
-    )
+        rf'|\b(?:cp|mv)\s+[^\s|;&]+\s+[^\n|;&]{{0,40}}?{file_alt}(?!\.?\w)')
 
 
 def _prose_modify_re(file_alt: str) -> str:
@@ -97,8 +95,7 @@ def _prose_modify_re(file_alt: str) -> str:
         rf'^\s*(?:[-*+]\s+|\d+[.)]\s+)?{MODIFY_VERB_RE}[^\n,]{{0,80}}?{file_alt}\b'
         rf'|(?:\byou\s+(?:must|should|need\s+to)\s+|\bplease\s+'
         rf'|\bmake\s+sure\s+(?:to\s+|you\s+)|\bbe\s+sure\s+to\s+)'
-        rf'{MODIFY_VERB_RE}[^\n,]{{0,80}}?{file_alt}\b'
-    )
+        rf'{MODIFY_VERB_RE}[^\n,]{{0,80}}?{file_alt}\b')
 
 
 def _content_contract_re(file_alt: str) -> str:
@@ -106,8 +103,7 @@ def _content_contract_re(file_alt: str) -> str:
     separable statically, so the tier is scored high (caution → confirmation), never critical."""
     return (
         rf'{file_alt}\b[^\n]{{0,40}}?\b(?:should|must|needs?\s+to)\s+'
-        rf'(?:contain|say|include|have|list)\b'
-    )
+        rf'(?:contain|say|include|have|list)\b')
 
 THREAT_PATTERNS = [
     # ── Exfiltration: shell commands leaking secrets ──
@@ -352,10 +348,7 @@ THREAT_PATTERNS = [
      "send_to_url", "high", "exfiltration", "instructs agent to send data to a URL"),
 ]
 
-_COMPILED_THREAT_PATTERNS = [
-    (re.compile(pattern, re.IGNORECASE), pid, severity, category, description)
-    for pattern, pid, severity, category, description in THREAT_PATTERNS
-]
+_COMPILED_THREAT_PATTERNS = [(re.compile(pattern, re.IGNORECASE), *rest) for pattern, *rest in THREAT_PATTERNS]
 
 # Structural limits for skill directories
 MAX_FILE_COUNT = 50       # skills shouldn't have 50+ files
@@ -393,13 +386,12 @@ def _compute_docstring_lines(lines: list) -> set:
     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):
-        was_in = in_docstring
-        counts = [line.count(marker) for marker in ('"""', "'''")]
-        in_docstring ^= sum(counts) % 2 == 1  # each odd marker count toggles; two odd counts cancel
-        if was_in or in_docstring or any(counts):
-            doc_lines.add(i + 1)
+    inside = False
+    for i, line in enumerate(lines, start=1):
+        was_in, counts = inside, [line.count(marker) for marker in ('"""', "'''")]
+        inside ^= sum(counts) % 2 == 1  # each odd marker count toggles; two odd counts cancel
+        if was_in or inside or any(counts):
+            doc_lines.add(i)
     return doc_lines
 
 
@@ -505,8 +497,7 @@ def scan_skill_cached(
 def should_allow_install(result: ScanResult, force: bool = False) -> Tuple[bool, str]:
     """``(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)]
+    decision = INSTALL_POLICY.get(result.trust_level, INSTALL_POLICY["community"])[VERDICT_INDEX.get(result.verdict, 2)]
     n = len(result.findings)
     hard_block = result.verdict == "dangerous" and result.trust_level in ("community", "trusted")
     if decision == "allow":
@@ -515,11 +506,8 @@ def should_allow_install(result: ScanResult, force: bool = False) -> Tuple[bool,
         return True, f"Force-installed despite {result.verdict} verdict ({n} findings)"
     if decision == "ask":
         return None, f"Requires confirmation ({result.trust_level} source + {result.verdict} verdict, {n} findings)"
-    if hard_block:
-        return False, (f"Blocked ({result.trust_level} source + dangerous verdict, {n} findings). "
-                       "--force does not override a dangerous verdict.")
-    return False, (f"Blocked ({result.trust_level} source + {result.verdict} verdict, {n} findings). "
-                   "Use --force to override.")
+    blocked = f"Blocked ({result.trust_level} source + {result.verdict} verdict, {n} findings). "
+    return False, blocked + ("--force does not override a dangerous verdict." if hard_block else "Use --force to override.")
 
 
 def format_scan_report(result: ScanResult) -> str:
@@ -533,8 +521,7 @@ def format_scan_report(result: ScanResult) -> str:
         lines.append("")
     allowed, reason = should_allow_install(result)
     status = "ALLOWED" if allowed is True else "NEEDS CONFIRMATION" if allowed is None else "BLOCKED"
-    lines.append(f"Decision: {status} — {reason}")
-    return "\n".join(lines)
+    return "\n".join(lines + [f"Decision: {status} — {reason}"])
 
 
 def _check_structure(skill_dir: Path, ignore=None) -> List[Finding]:
@@ -560,9 +547,10 @@ def _check_structure(skill_dir: Path, ignore=None) -> List[Finding]:
                 add("broken_symlink", "medium", "traversal", rel, "broken symlink", "broken or circular symlink")
             continue
         try:
-            size = f.stat().st_size
+            st = f.stat()
         except OSError:
             continue
+        size = st.st_size
         total_size += size
         if size > MAX_SINGLE_FILE_KB * 1024:
             add("oversized_file", "medium", "structural", rel, f"{size // 1024}KB",
@@ -571,7 +559,7 @@ def _check_structure(skill_dir: Path, ignore=None) -> List[Finding]:
         if ext in SUSPICIOUS_BINARY_EXTENSIONS:
             add("binary_file", "critical", "structural", rel, f"binary: {ext}",
                 f"binary/executable file ({ext}) should not be in a skill")
-        if ext not in _SCRIPT_EXTENSIONS and f.stat().st_mode & 0o111:
+        if ext not in _SCRIPT_EXTENSIONS and st.st_mode & 0o111:
             add("unexpected_executable", "medium", "structural", rel, "executable bit set",
                 "file has executable permission but is not a recognized script type")
     if file_count > MAX_FILE_COUNT:
@@ -585,8 +573,6 @@ def _check_structure(skill_dir: Path, ignore=None) -> List[Finding]:
 
 # `.skillignore` is Hermes-native; `.clawhubignore` is honored for skills published through ClawHub.
 _SKILL_IGNORE_FILENAMES = (".skillignore", ".clawhubignore")
-_ALWAYS_IGNORED_NAMES = set(_SKILL_IGNORE_FILENAMES)
-_NEVER_IGNORABLE = {"SKILL.md"}
 
 
 def _load_skill_ignore(skill_dir: Path):
@@ -605,9 +591,9 @@ def _load_skill_ignore(skill_dir: Path):
         rel_posix = Path(rel).as_posix()
         segs = rel_posix.split("/")
         base = segs[-1]
-        if base in _NEVER_IGNORABLE:
+        if base == "SKILL.md":
             return False
-        if base in _ALWAYS_IGNORED_NAMES:
+        if base in _SKILL_IGNORE_FILENAMES:
             return True
         for pat in patterns:
             anchored = pat.startswith("/")