hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 29 Packages Projects Releases Wiki Activity

refactor(tools): skill linter checks as generators, skill_manage arg-shape table, suppress() collapses

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:43:10 -07:00
parent 9832b0800d
commit b187d4e547
5 changed files with 212 additions and 362 deletions
Download Patch File Download Diff File Expand all files Collapse all files

109
tools/skill_ledger.py
View File

@@ -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/<name>/``, 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/<name>/``, 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]] = []

270
tools/skill_linter.py
View File

@@ -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]:

40
tools/skill_manager_batch.py
View File

@@ -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)

31
tools/skill_manager_guards.py
View File

@@ -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 <name>` 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(

124
tools/skill_manager_tool.py
View File

@@ -10,6 +10,7 @@ place. Layout: ``<skills>/[category/]<skill>/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")),
)