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

refactor(tools): trim skill_manage stack — dead linter CLI/format helpers, merged path resolvers, compact layout

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:23:24 -07:00
parent 113f04616b
commit 71dadfeb99
6 changed files with 264 additions and 686 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
tests/tools/test_skill_linter.py
View File

@@ -7,8 +7,6 @@ import pytest
from tools.skill_linter import (
ERROR,
WARNING,
format_findings,
has_errors,
lint_content,
lint_skill,
)
@@ -105,7 +103,7 @@ def test_bad_name_format_is_error():
content = CLEAN.replace("name: my-skill", "name: My_Skill!")
findings = lint_content(content)
assert "name-format" in _rules(findings)
assert has_errors(findings)
assert any(f.severity == ERROR for f in findings)
def test_name_dir_mismatch_is_error(tmp_path):
@@ -113,7 +111,7 @@ def test_name_dir_mismatch_is_error(tmp_path):
skill_dir.mkdir()
findings = lint_content(CLEAN, skill_dir=skill_dir) # name is my-skill
assert "name-dir-mismatch" in _rules(findings)
assert has_errors(findings)
assert any(f.severity == ERROR for f in findings)
def test_dangling_reference_link_flagged(tmp_path):
@@ -183,7 +181,6 @@ def test_author_caps_warned():
assert "author-caps" in _rules(findings)
def test_format_findings_renders():
def test_findings_carry_rule_and_severity():
findings = lint_content(CLEAN.replace("name: my-skill", "name: BAD"))
out = format_findings(findings)
assert "name-format" in out
assert any(f.rule == "name-format" and f.severity == ERROR for f in findings)

175
tools/skill_ledger.py
View File

@@ -1,23 +1,12 @@
"""Per-mutation skill audit ledger + single-edit rollback.
Every skill mutation — any actor — appends one JSONL entry to
``~/.hermes/skills/.curator_ledger.jsonl`` with before/after file manifests
whose contents are stored content-addressed (sha256-deduped) under
``~/.hermes/.curator_backups/blobs/``.
Design decisions (Teknium-approved):
- JSONL, not the state DB: durable, human-greppable, survives DB resets.
- Covers ALL actors (``curator`` / ``agent`` / ``user``). The curator
invariant (never hard-delete autonomously) applies only to autonomous
actors; user deletes stay hard-delete but are ledgered and recoverable via
``hermes curator rollback <entry-id>``.
- Per-file content-addressed blobs, not tarballs: a mutation usually touches
one file, and identical content dedupes to one blob.
The ledger is TELEMETRY, NOT A GATE: a ledger failure must never block the
mutation it describes — every public write path swallows and logs. The one
exception is ``rollback_entry``, which FAILS CLOSED when its own pre-rollback
safety capture fails (consistent with agent/curator_backup.py).
Every skill mutation (any actor) appends one JSONL entry to
``~/.hermes/skills/.curator_ledger.jsonl`` with before/after file manifests whose
contents are stored content-addressed (sha256-deduped) under
``~/.hermes/.curator_backups/blobs/``. JSONL, not the state DB: durable,
human-greppable, survives DB resets. The ledger is TELEMETRY, NOT A GATE: every
public write path swallows and logs. The one exception is ``rollback_entry``,
which FAILS CLOSED when its own pre-rollback safety capture fails.
"""
from __future__ import annotations
@@ -38,10 +27,6 @@ from hermes_constants import get_hermes_home
logger = logging.getLogger(__name__)
ACTOR_CURATOR = "curator"
ACTOR_AGENT = "agent"
ACTOR_USER = "user"
# Snapshot-id shape used by agent.curator_backup (duplicated so the ledger can
# read the newest skills.tar.gz without importing the backup stack).
_BACKUP_ID_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}-\d{2}-\d{2}Z(-\d{2})?$")
@@ -51,13 +36,12 @@ _ARCHIVE_TS_SUFFIX_RE = re.compile(r"^(.+)-\d{14}$")
# have re-homed support files out of the tree first, so a disk-only capture
# would make rollback restore a hollow skill.
_PACKAGE_RESTORE_ACTIONS = frozenset({"delete", "archive", "purge"})
_VALID_ACTORS = {ACTOR_CURATOR, ACTOR_AGENT, ACTOR_USER}
_VALID_ACTORS = {"curator", "agent", "user"}
_NON_PACKAGE_TOPS = {".curator_backups", ".hub", ".archive"}
# Explicit actor override: the CLI sets "user", the curator walk sets "curator".
_actor_override: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar(
"skill_ledger_actor", default=None
)
"skill_ledger_actor", default=None)
def set_ledger_actor(actor: Optional[str]) -> contextvars.Token:
@@ -76,18 +60,13 @@ def derive_actor() -> str:
return override
try:
from tools.skill_provenance import is_background_review
if is_background_review():
return ACTOR_CURATOR
return "curator"
except Exception:
pass
return ACTOR_AGENT
return "agent"
# ---------------------------------------------------------------------------
# Paths + config gate
# ---------------------------------------------------------------------------
def ledger_path() -> Path:
return get_hermes_home() / "skills" / ".curator_ledger.jsonl"
@@ -101,11 +80,10 @@ def _skills_dir() -> Path:
def ledger_enabled() -> bool:
"""Config gate ``skills.ledger`` (default True). Lazy import so this
module stays importable without the CLI config layer."""
"""Config gate ``skills.ledger`` (default True); lazy import keeps the module
importable without the CLI config layer."""
try:
from hermes_cli.config import cfg_get, load_config
return bool(cfg_get(load_config(), "skills", "ledger", default=True))
except Exception as e: # pragma: no cover — best-effort config read
logger.debug("skill_ledger: config read failed (%s); defaulting on", e)
@@ -133,9 +111,7 @@ def _is_within(root: Path, path: Path) -> bool:
return False
# ---------------------------------------------------------------------------
# Blob store (content-addressed, deduped)
# ---------------------------------------------------------------------------
# --- Blob store (content-addressed, deduped) ---------------------------------
def _store_blob(data: bytes) -> str:
"""Write *data* keyed by sha256 (existing blob left alone). Returns the hash."""
@@ -160,11 +136,7 @@ def read_blob(sha256: str) -> Optional[bytes]:
return None
def snapshot_paths(
root: Optional[Path],
*,
complete_package: bool = False,
) -> List[Dict[str, str]]:
def snapshot_paths(root: Optional[Path], *, complete_package: bool = False) -> List[Dict[str, str]]:
"""Capture {path, sha256} for every file under *root*, storing each as a blob.
Empty when root is None/missing. Raises on I/O failure — callers decide
@@ -188,9 +160,7 @@ def snapshot_paths(
return out
# ---------------------------------------------------------------------------
# Package-completeness fill from the newest curator backup
# ---------------------------------------------------------------------------
# --- Package-completeness fill from the newest curator backup -----------------
def _package_rel(root: Path) -> Optional[str]:
"""Relative POSIX path of a skill dir under ``skills/``; None when outside
@@ -215,10 +185,8 @@ def _skill_md_parent(items: Optional[List[Dict[str, str]]]) -> Optional[Path]:
def package_prefixes(
root: Optional[Path] = None,
skill: Optional[str] = None,
before: Optional[List[Dict[str, str]]] = None,
) -> List[str]:
root: Optional[Path] = None, skill: Optional[str] = None,
before: Optional[List[Dict[str, str]]] = None) -> List[str]:
"""Tar member prefixes that belong to this skill's package: its live
location under ``skills/``, the package parent recorded in the before-state
SKILL.md path (for rollback fills where *root* is gone), the bare skill
@@ -247,8 +215,7 @@ def _latest_skills_tarball() -> Optional[Path]:
except OSError:
return None
candidates = [
child / "skills.tar.gz"
for child in children
child / "skills.tar.gz" for child in children
if child.is_dir() and _BACKUP_ID_RE.match(child.name) and (child / "skills.tar.gz").is_file()
]
if not candidates:
@@ -288,21 +255,17 @@ def _read_package_files_from_latest_backup(prefixes: List[str]) -> Dict[str, byt
def fill_snapshot_from_curator_backup(
root: Optional[Path],
existing: Optional[List[Dict[str, str]]] = None,
*,
skill: Optional[str] = None,
) -> List[Dict[str, str]]:
root: Optional[Path], existing: Optional[List[Dict[str, str]]] = None, *,
skill: Optional[str] = None) -> List[Dict[str, str]]:
"""Union missing skill-package files from the newest curator snapshot.
Completeness fill, not a gate: failures are swallowed and *existing* is
returned unchanged; the backup only fills paths ABSENT from it.
Filled files are addressed where the rollback must restore them: under
*root* when known (for purge that is ``.archive/<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.
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.
"""
out = list(existing or [])
prefixes = package_prefixes(root, skill, out)
@@ -342,18 +305,12 @@ def fill_snapshot_from_curator_backup(
return out
# ---------------------------------------------------------------------------
# Append + read
# ---------------------------------------------------------------------------
# --- Append + read ------------------------------------------------------------
def append_entry(
action: str,
skill: str,
before: Optional[List[Dict[str, str]]] = None,
after: Optional[List[Dict[str, str]]] = None,
actor: Optional[str] = None,
evidence: Optional[Dict[str, Any]] = None,
) -> Optional[str]:
action: str, skill: str, before: Optional[List[Dict[str, str]]] = None,
after: Optional[List[Dict[str, str]]] = None, actor: Optional[str] = None,
evidence: Optional[Dict[str, Any]] = None) -> Optional[str]:
"""Append one ledger entry. Returns the entry id, or None when the
ledger is disabled or the write failed (never raises)."""
if not ledger_enabled():
@@ -367,8 +324,7 @@ def append_entry(
"skill": skill,
"evidence": evidence or {},
"before": before or [],
"after": after or [],
}
"after": after or []}
path = ledger_path()
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "a", encoding="utf-8") as fh:
@@ -380,14 +336,9 @@ def append_entry(
def record_mutation(
action: str,
skill: str,
before_root: Optional[Path] = None,
before: Optional[List[Dict[str, str]]] = None,
after_root: Optional[Path] = None,
actor: Optional[str] = None,
evidence: Optional[Dict[str, Any]] = None,
) -> Optional[str]:
action: str, skill: str, before_root: Optional[Path] = None,
before: Optional[List[Dict[str, str]]] = None, after_root: Optional[Path] = None,
actor: Optional[str] = None, evidence: Optional[Dict[str, Any]] = None) -> Optional[str]:
"""One-stop hook for mutation call sites: capture after-state from
*after_root* (pre-captured *before* list, or capture from *before_root*)
and append. NEVER raises and never blocks the mutation.
@@ -404,18 +355,14 @@ def record_mutation(
before = fill_snapshot_from_curator_backup(before_root, before, skill=skill)
after = snapshot_paths(after_root)
return append_entry(
action, skill, before=before, after=after, actor=actor, evidence=evidence
)
action, skill, before=before, after=after, actor=actor, evidence=evidence)
except Exception as e:
logger.warning("skill_ledger: record_mutation failed (%s) — mutation unaffected", e)
return None
def capture_before(
root: Optional[Path],
*,
complete_package: bool = False,
skill: Optional[str] = None,
root: Optional[Path], *, complete_package: bool = False, skill: Optional[str] = None,
) -> Optional[List[Dict[str, str]]]:
"""Best-effort pre-mutation capture; None on failure or when disabled
(callers pass the result straight to record_mutation). Use
@@ -432,9 +379,7 @@ def capture_before(
return None
def list_entries(
skill: Optional[str] = None, limit: Optional[int] = None
) -> List[Dict[str, Any]]:
def list_entries(skill: Optional[str] = None, limit: Optional[int] = None) -> List[Dict[str, Any]]:
"""Read the ledger, newest first. Malformed lines are skipped."""
path = ledger_path()
if not path.exists():
@@ -468,9 +413,7 @@ def get_entry(entry_id: str) -> Optional[Dict[str, Any]]:
return next((row for row in list_entries() if row.get("id") == entry_id), None)
# ---------------------------------------------------------------------------
# Single-edit rollback
# ---------------------------------------------------------------------------
# --- Single-edit rollback -----------------------------------------------------
def _validate_entry_paths(entry: Dict[str, Any]) -> Optional[str]:
"""All paths in an entry must live under HERMES_HOME — a hand-edited
@@ -509,8 +452,7 @@ def rollback_entry(entry_id: str) -> Tuple[bool, str]:
# added, and the filled set is re-validated against HERMES_HOME.
if entry.get("action") in _PACKAGE_RESTORE_ACTIONS:
before = fill_snapshot_from_curator_backup(
_skill_md_parent(before), before, skill=str(entry.get("skill") or "") or None
)
_skill_md_parent(before), before, skill=str(entry.get("skill") or "") or None)
path_err = _validate_entry_paths({**entry, "before": before, "after": after})
if path_err:
return False, f"refusing rollback: {path_err}"
@@ -518,10 +460,8 @@ def rollback_entry(entry_id: str) -> Tuple[bool, str]:
# Pre-check every blob we need so we never fail mid-restore.
for item in before:
if read_blob(str(item.get("sha256", ""))) is None:
return False, (
f"missing blob {item.get('sha256')} for {item.get('path')}; "
"rollback aborted, nothing was changed"
)
return False, (f"missing blob {item.get('sha256')} for {item.get('path')}; "
"rollback aborted, nothing was changed")
# Touched paths = union of before/after. Capture their CURRENT state as
# the safety entry so the rollback itself is undoable. FAIL CLOSED.
@@ -533,22 +473,14 @@ def rollback_entry(entry_id: str) -> Tuple[bool, str]:
if fp.is_file():
safety_before.append({"path": p, "sha256": _store_blob(fp.read_bytes())})
safety_id = append_entry(
"pre-rollback",
entry.get("skill", "?"),
before=safety_before,
after=safety_before,
evidence={"rollback_target": entry_id},
)
"pre-rollback", entry.get("skill", "?"), before=safety_before, after=safety_before,
evidence={"rollback_target": entry_id})
except Exception as e:
return False, (
f"pre-rollback safety capture failed ({e}); rollback aborted and "
"current skills were not changed"
)
return False, (f"pre-rollback safety capture failed ({e}); rollback aborted and "
"current skills were not changed")
if safety_id is None:
return False, (
"pre-rollback safety capture failed (ledger disabled or "
"unwritable); rollback aborted and current skills were not changed"
)
return False, ("pre-rollback safety capture failed (ledger disabled or "
"unwritable); rollback aborted and current skills were not changed")
# Restore: write every before-file, remove files the mutation created.
before_paths = {str(i["path"]) for i in before}
@@ -573,14 +505,9 @@ def rollback_entry(entry_id: str) -> Tuple[bool, str]:
logger.warning("skill_ledger: could not remove %s during rollback: %s", p, e)
append_entry(
"rollback",
entry.get("skill", "?"),
before=safety_before,
after=before,
evidence={"rollback_target": entry_id, "restored": restored, "removed": removed},
)
"rollback", entry.get("skill", "?"), before=safety_before, after=before,
evidence={"rollback_target": entry_id, "restored": restored, "removed": removed})
return True, (
f"rolled back entry {entry_id} ({entry.get('action')} on "
f"'{entry.get('skill')}'): {restored} file(s) restored, {removed} removed. "
f"Safety entry {safety_id} captured the pre-rollback state."
)
f"Safety entry {safety_id} captured the pre-rollback state.")

254
tools/skill_linter.py
View File

@@ -1,18 +1,11 @@
"""Structural + convention linter for SKILL.md files.
The hard *validator* (``tools/skill_manager_tool.py::_validate_frontmatter``)
blocks the non-negotiables (fence, YAML mapping, name/description, size caps).
This module is the softer companion encoding the CONTRIBUTING.md "Skill
authoring standards" that otherwise only a human reviewer catches: shell
utilities named instead of native tools, missing author/license/metadata,
``name`` != directory, dangling ``references/`` links, marketing words,
``platforms:`` gating vs POSIX-only scripts, forbidden scaffolding files.
Contract: findings are advisory — ``lint_skill`` returns ``LintFinding`` rows and
the caller decides what blocks. Pure functions (I/O limited to the files pointed
at) so CI, the ``skill_manage`` create path and local runs share one impl.
Frontmatter parsing is delegated to ``agent.skill_utils`` so BOM handling and
the prompt description budget stay in one place.
The hard validator (``skill_manager_tool._validate_frontmatter``) blocks the
non-negotiables; this is the advisory companion encoding the CONTRIBUTING.md
"Skill authoring standards" a human reviewer would otherwise catch. Findings
never block by themselves — ``lint_skill`` returns ``LintFinding`` rows and the
caller decides. Frontmatter parsing is delegated to ``agent.skill_utils`` so
BOM handling and the prompt description budget stay in one place.
"""
from __future__ import annotations
@@ -22,39 +15,17 @@ from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, List, Optional
from agent.skill_utils import (
SKILL_PROMPT_DESC_LIMIT,
parse_frontmatter,
)
# ── Rule data ────────────────────────────────────────────────────────────────
from agent.skill_utils import SKILL_PROMPT_DESC_LIMIT, parse_frontmatter
# Shell utilities already wrapped as native tools; naming them in prose steers
# the model to a raw shell call. banned token -> native tool to name instead.
_SHELL_UTIL_TO_TOOL: Dict[str, str] = {
"grep": "search_files",
"rg": "search_files",
"cat": "read_file",
"head": "read_file",
"tail": "read_file",
"sed": "patch",
"awk": "patch",
"find": "search_files (target='files')",
"ls": "search_files (target='files')",
}
# Marketing words the description must not contain.
"grep": "search_files", "rg": "search_files", "cat": "read_file", "head": "read_file",
"tail": "read_file", "sed": "patch", "awk": "patch",
"find": "search_files (target='files')", "ls": "search_files (target='files')"}
_MARKETING_WORDS = (
"powerful",
"comprehensive",
"seamless",
"advanced",
"cutting-edge",
"state-of-the-art",
"revolutionary",
"robust",
)
"powerful", "comprehensive", "seamless", "advanced", "cutting-edge", "state-of-the-art",
"revolutionary", "robust")
# POSIX-only primitives that require ``platforms:`` when a bundled script uses
# them. Detected in scripts/, not in prose.
_POSIX_PRIMITIVES = (
@@ -65,19 +36,9 @@ _POSIX_PRIMITIVES = (
"osascript",
"/proc/",
"apt-get",
"systemctl",
)
"systemctl")
# Scaffolding files a skill should not ship (noise, not skill content).
_FORBIDDEN_FILES = (
"README.md",
"CHANGELOG.md",
"install.sh",
".env",
".env.example",
".gitignore",
)
_FORBIDDEN_FILES = ("README.md", "CHANGELOG.md", "install.sh", ".env", ".env.example", ".gitignore")
# Presence of the load-bearing section is checked, not exact ordering, so the
# linter is not a change-detector.
_EXPECTED_SECTIONS = ("When to Use", "When to use")
@@ -94,10 +55,6 @@ class LintFinding:
rule: str
message: str
def format(self) -> str:
badge = "✗" if self.severity == ERROR else "⚠"
return f"{badge} [{self.rule}] {self.message}"
def _err(rule: str, message: str) -> LintFinding:
return LintFinding(ERROR, rule, message)
@@ -107,32 +64,23 @@ def _warn(rule: str, message: str) -> LintFinding:
return LintFinding(WARNING, rule, message)
# ── Individual checks ────────────────────────────────────────────────────────
def _check_name_matches_dir(
frontmatter: Dict[str, Any], skill_dir: Optional[Path]
frontmatter: Dict[str, Any], skill_dir: Optional[Path],
) -> List[LintFinding]:
if skill_dir is None:
return []
name = str(frontmatter.get("name", "")).strip()
if name and name != skill_dir.name:
return [_err(
"name-dir-mismatch",
f"frontmatter name '{name}' does not match directory "
f"'{skill_dir.name}'; they must be identical.",
)]
return [_err("name-dir-mismatch", f"frontmatter name '{name}' does not match directory "
f"'{skill_dir.name}'; they must be identical.")]
return []
def _check_name_format(frontmatter: Dict[str, Any]) -> List[LintFinding]:
name = str(frontmatter.get("name", "")).strip()
if name and not re.fullmatch(r"[a-z0-9][a-z0-9_-]*", name):
return [_err(
"name-format",
f"name '{name}' must be lowercase letters, digits, hyphens, "
f"and underscores only.",
)]
return [_err("name-format", f"name '{name}' must be lowercase letters, digits, hyphens, "
f"and underscores only.")]
return []
@@ -146,8 +94,8 @@ def _check_description(frontmatter: Dict[str, Any]) -> List[LintFinding]:
if len(desc) > SKILL_PROMPT_DESC_LIMIT:
findings.append(_warn(
"description-length",
f"description is {len(desc)} chars; the skill index truncates "
f"past {SKILL_PROMPT_DESC_LIMIT} chars + '...', losing routing "
f"description is {len(desc)} chars; the skill index truncates past "
f"{SKILL_PROMPT_DESC_LIMIT} chars + '...', losing routing "
f"signal. Keep it to one sentence.",
))
lower = desc.lower()
@@ -155,9 +103,7 @@ def _check_description(frontmatter: Dict[str, Any]) -> List[LintFinding]:
if hits:
findings.append(_warn(
"description-marketing",
f"description contains marketing words {hits}; state the "
f"capability, not adjectives.",
))
f"description contains marketing words {hits}; state the capability, not adjectives."))
return findings
@@ -166,26 +112,20 @@ def _check_metadata_block(frontmatter: Dict[str, Any]) -> List[LintFinding]:
for key in ("version", "author", "license"):
if key not in frontmatter:
findings.append(_warn(
"missing-metadata",
f"frontmatter is missing '{key}'; every peer skill has it.",
))
"missing-metadata", f"frontmatter is missing '{key}'; every peer skill has it."))
meta = frontmatter.get("metadata")
hermes_meta = meta.get("hermes") if isinstance(meta, dict) else None
if not isinstance(hermes_meta, dict):
findings.append(_warn(
"missing-metadata",
"frontmatter is missing metadata.hermes.{tags, related_skills}.",
))
"missing-metadata", "frontmatter is missing metadata.hermes.{tags, related_skills}."))
elif "tags" not in hermes_meta:
findings.append(_warn("missing-metadata", "metadata.hermes.tags is missing."))
author = str(frontmatter.get("author", ""))
if author and author.strip().lower() in ("hermes", "agent", "hermes agent") and (
author != "Hermes Agent"
):
author != "Hermes Agent"):
findings.append(_warn(
"author-caps",
f"author '{author}' should be 'Hermes Agent' (proper caps) "
f"or a real contributor name.",
f"author '{author}' should be 'Hermes Agent' (proper caps) or a real contributor name.",
))
return findings
@@ -195,23 +135,16 @@ def _check_shell_utilities(body: str) -> List[LintFinding]:
prose = _strip_code_blocks(body)
# Only backtick-wrapped mentions: bare words in sentences are too noisy.
return [
_warn(
"shell-utility-reference",
f"prose references `{util}`; name the native tool "
f"`{tool}` instead.",
)
_warn("shell-utility-reference",
f"prose references `{util}`; name the native tool `{tool}` instead.")
for util, tool in _SHELL_UTIL_TO_TOOL.items()
if re.search(rf"`{re.escape(util)}`", prose)
]
if re.search(rf"`{re.escape(util)}`", prose)]
def _check_sections(body: str) -> List[LintFinding]:
if not any(re.search(rf"^#+\s+{re.escape(s)}", body, re.M) for s in _EXPECTED_SECTIONS):
return [_warn(
"missing-section",
"no '## When to Use' section found; skills need explicit "
"trigger conditions near the top.",
)]
return [_warn("missing-section", "no '## When to Use' section found; skills need explicit "
"trigger conditions near the top.")]
return []
@@ -231,16 +164,13 @@ def _check_reference_links(body: str, skill_dir: Optional[Path]) -> List[LintFin
if "*" in rel or rel.endswith("/"): # placeholders / globs
continue
if not (skill_dir / rel).exists():
findings.append(_warn(
"dangling-reference",
f"body references '{rel}' but that file does not exist "
f"in the skill directory.",
))
findings.append(_warn("dangling-reference", f"body references '{rel}' but that file "
f"does not exist in the skill directory."))
return findings
def _check_platforms_gating(
frontmatter: Dict[str, Any], skill_dir: Optional[Path]
frontmatter: Dict[str, Any], skill_dir: Optional[Path],
) -> List[LintFinding]:
"""If bundled scripts use POSIX-only primitives, require platforms:."""
if skill_dir is None or frontmatter.get("platforms"):
@@ -263,10 +193,8 @@ def _check_platforms_gating(
detail = "; ".join(f"{k}: {v}" for k, v in offenders.items())
return [_warn(
"platforms-gating",
f"scripts use POSIX-only primitives ({detail}) but no "
f"'platforms:' frontmatter is declared. Fix cross-platform or "
f"gate with platforms: [linux, macos].",
)]
f"scripts use POSIX-only primitives ({detail}) but no 'platforms:' frontmatter is "
f"declared. Fix cross-platform or gate with platforms: [linux, macos].")]
return []
@@ -274,14 +202,10 @@ def _check_forbidden_files(skill_dir: Optional[Path]) -> List[LintFinding]:
if skill_dir is None:
return []
return [
_warn(
"forbidden-file",
f"skill ships '{fname}'; skills should not include "
f"scaffolding/config files.",
)
_warn("forbidden-file",
f"skill ships '{fname}'; skills should not include scaffolding/config files.")
for fname in _FORBIDDEN_FILES
if (skill_dir / fname).exists()
]
if (skill_dir / fname).exists()]
def _check_platform_list_valid(frontmatter: Dict[str, Any]) -> List[LintFinding]:
@@ -292,11 +216,8 @@ def _check_platform_list_valid(frontmatter: Dict[str, Any]) -> List[LintFinding]
items = platforms if isinstance(platforms, list) else [platforms]
bad = [p for p in items if str(p).lower() not in valid]
if bad:
return [_warn(
"platforms-value",
f"platforms contains unrecognized value(s) {bad}; expected a "
f"subset of {sorted(valid)}.",
)]
return [_warn("platforms-value", f"platforms contains unrecognized value(s) {bad}; "
f"expected a subset of {sorted(valid)}.")]
return []
@@ -305,31 +226,25 @@ def _strip_code_blocks(body: str) -> str:
return re.sub(r"```.*?```", "", body, flags=re.S)
# ── Public API ───────────────────────────────────────────────────────────────
def lint_content(
content: str, *, skill_dir: Optional[Path] = None
) -> List[LintFinding]:
def lint_content(content: str, *, skill_dir: Optional[Path] = None) -> List[LintFinding]:
"""Lint raw SKILL.md *content*.
``skill_dir`` enables on-disk checks (name/dir match, dangling links,
POSIX gating, forbidden files); without it only content checks run,
which is what the create path needs before the file exists.
``skill_dir`` enables on-disk checks (name/dir match, dangling links, POSIX
gating, forbidden files); without it only content checks run, which is what
the create path needs before the file exists.
"""
frontmatter, body = parse_frontmatter(content)
findings: List[LintFinding] = []
findings += _check_name_format(frontmatter)
findings += _check_name_matches_dir(frontmatter, skill_dir)
findings += _check_description(frontmatter)
findings += _check_metadata_block(frontmatter)
findings += _check_platform_list_valid(frontmatter)
findings += _check_shell_utilities(body)
findings += _check_sections(body)
findings += _check_reference_links(body, skill_dir)
findings += _check_platforms_gating(frontmatter, skill_dir)
findings += _check_forbidden_files(skill_dir)
return findings
return (
_check_name_format(frontmatter)
+ _check_name_matches_dir(frontmatter, skill_dir)
+ _check_description(frontmatter)
+ _check_metadata_block(frontmatter)
+ _check_platform_list_valid(frontmatter)
+ _check_shell_utilities(body)
+ _check_sections(body)
+ _check_reference_links(body, skill_dir)
+ _check_platforms_gating(frontmatter, skill_dir)
+ _check_forbidden_files(skill_dir))
def lint_skill(skill_md_path: Path) -> List[LintFinding]:
@@ -337,60 +252,3 @@ def lint_skill(skill_md_path: Path) -> List[LintFinding]:
skill_md_path = Path(skill_md_path)
content = skill_md_path.read_text(encoding="utf-8", errors="ignore")
return lint_content(content, skill_dir=skill_md_path.parent)
def format_findings(findings: List[LintFinding]) -> str:
"""Render findings as a newline-joined human-readable block."""
return "\n".join(f.format() for f in findings)
def has_errors(findings: List[LintFinding]) -> bool:
return any(f.severity == ERROR for f in findings)
def _main(argv: Optional[List[str]] = None) -> int:
"""CLI: ``python -m tools.skill_linter <SKILL.md | skill-dir> ...``
Exit 1 only on ERROR-severity findings (WARNING-only exits 0) so CI can
gate on structural breakage without failing on advisory nits.
"""
import sys
args = argv if argv is not None else sys.argv[1:]
if not args:
print("usage: python -m tools.skill_linter <SKILL.md | skill-dir> ...")
return 2
targets: List[Path] = []
for arg in args:
p = Path(arg)
if p.is_dir():
targets.extend(sorted(p.rglob("SKILL.md")))
elif p.name == "SKILL.md" and p.is_file():
targets.append(p)
else:
print(f"skip (not a SKILL.md or dir): {arg}")
any_error = False
total = 0
for skill_md in targets:
findings = lint_skill(skill_md)
if not findings:
continue
total += len(findings)
if has_errors(findings):
any_error = True
print(f"\n{skill_md.parent.name} ({skill_md}):")
print(format_findings(findings))
if total == 0:
print(f"All {len(targets)} skill(s) clean.")
else:
print(f"\n{total} finding(s) across {len(targets)} skill(s).")
return 1 if any_error else 0
if __name__ == "__main__":
import sys
sys.exit(_main())

77
tools/skill_manager_batch.py
View File

@@ -19,9 +19,7 @@ _BATCH_MAX_OPS = 20
def _norm_target(op) -> str:
fp = (op.get("file_path") or "").strip()
if not fp:
return "SKILL.md"
return posixpath.normpath(fp.lstrip("/"))
return posixpath.normpath(fp.lstrip("/")) if fp else "SKILL.md"
def _validate_batch_ops(operations, default_name, tool_error):
@@ -38,18 +36,15 @@ def _validate_batch_ops(operations, default_name, tool_error):
f"operations[{i}]: unknown action '{act}'. "
f"Batchable: {', '.join(sorted(_BATCH_OP_ACTIONS))}; "
"delete must be sole.",
success=False,
)
success=False)
nm = op.get("name") or default_name
if not nm:
return None, tool_error(f"operations[{i}] needs a 'name' (the skill it targets).", success=False)
names.append(nm)
if act == "create" and nm in names[:-1]:
return None, tool_error(
f"operations[{i}]: create for '{nm}' must precede that "
"skill's other ops.",
success=False,
)
f"operations[{i}]: create for '{nm}' must precede that skill's other ops.",
success=False)
preflight = _background_review_preflight(act, nm)
if preflight is not None:
return None, json.dumps(preflight, ensure_ascii=False)
@@ -75,8 +70,7 @@ def _validate_batch_ops(operations, default_name, tool_error):
"op would silently discard its work. One destructive op "
"(write_file/remove_file/full rewrite) per file per batch; "
"put it first, or fold the change in. Patch chains are fine.",
success=False,
)
success=False)
touched_files.add(key)
return names, None
@@ -147,29 +141,20 @@ def _rollback(snapshots, find_skill):
except Exception as exc: # noqa: BLE001
notes.append(
f"ROLLBACK FAILED for '{nm}' ({exc}); snapshot preserved at '{snap}'"
if snap is not None
else f"ROLLBACK FAILED for '{nm}' ({exc})"
)
if snap is not None else f"ROLLBACK FAILED for '{nm}' ({exc})")
return ("; ".join(notes) if notes else "all touched skills rolled back"), bool(notes)
def _skill_manage_batch(
operations,
default_name: str = None,
task_id: str = None,
session_id: str = None,
) -> str:
def _skill_manage_batch(operations, default_name: str = None, task_id: str = None,
session_id: str = None) -> str:
"""Apply a sequence of operations atomically (memory-tool pattern).
Each op carries its own ``name`` and ``action``. Every touched skill is
snapshotted before any op runs; any failure rolls ALL touched skills back
(skills the batch created are removed).
Rules: ``delete`` only as the SOLE op (its recoverable-archive path doesn't
compose with rollback) and is routed to the single-op handler, preserving
absorbed_into/archive semantics; ``create`` must precede that skill's other
ops; the same-file clobber guard rejects silently-lost work.
``default_name`` is the legacy top-level ``name`` fallback (staged replay).
Every touched skill is snapshotted before any op runs; any failure rolls ALL
touched skills back (skills the batch created are removed). ``delete`` is
only legal as the SOLE op (its recoverable-archive path doesn't compose with
rollback) and is routed to the single-op handler, preserving
absorbed_into/archive semantics. ``default_name`` is the legacy top-level
``name`` fallback (staged replay).
"""
from tools import skill_manager_tool as _smt
from tools.registry import tool_error
@@ -183,19 +168,14 @@ def _skill_manage_batch(
return tool_error(
"delete must be the SOLE op in its call — it doesn't "
"compose with other ops' rollback.",
success=False,
)
success=False)
op = operations[0]
nm = op.get("name") or default_name
if not nm:
return tool_error("operations[0] (delete) needs a 'name'.", success=False)
return _smt.skill_manage(
action="delete",
name=nm,
absorbed_into=op.get("absorbed_into"),
task_id=task_id,
session_id=session_id,
)
action="delete", name=nm, absorbed_into=op.get("absorbed_into"),
task_id=task_id, session_id=session_id)
names, err = _validate_batch_ops(operations, default_name, tool_error)
if err is not None:
@@ -220,8 +200,7 @@ def _skill_manage_batch(
try:
for i, op in enumerate(operations):
raw = _smt._skill_manage_from(
{**op, "name": names[i]}, task_id=task_id, session_id=session_id,
)
{**op, "name": names[i]}, task_id=task_id, session_id=session_id)
try:
parsed = json.loads(raw)
except Exception: # noqa: BLE001
@@ -232,11 +211,9 @@ def _skill_manage_batch(
"success": False,
"error": (
f"operations[{i}] ({op['action']} on '{names[i]}') failed: "
f"{parsed.get('error', 'unknown error')} — batch aborted, {note}."
),
f"{parsed.get('error', 'unknown error')} — batch aborted, {note}."),
"failed_index": i,
"completed_before_failure": i,
}
"completed_before_failure": i}
# Carry the failing op's teaching payload (patch's file_preview /
# fuzzy-match hints) through — without it the model recovers blind.
for k, v in parsed.items():
@@ -244,21 +221,15 @@ def _skill_manage_batch(
fail.setdefault(k, v)
return json.dumps(fail, ensure_ascii=False)
results.append({"name": names[i], "action": op["action"],
"file_path": op.get("file_path"),
"success": True})
"file_path": op.get("file_path"), "success": True})
finally:
_smt._skill_gate_bypass.reset(token)
if rollback_failed:
# Keep the snapshots so the operator can still recover by hand.
logger.warning(
"skill_manage batch rollback failed, snapshots kept at %s",
snap_root,
)
logger.warning("skill_manage batch rollback failed, snapshots kept at %s", snap_root)
else:
shutil.rmtree(snap_root, ignore_errors=True)
return json.dumps(
{"success": True, "operations_applied": len(results),
"results": results},
ensure_ascii=False,
)
{"success": True, "operations_applied": len(results), "results": results},
ensure_ascii=False)

156
tools/skill_manager_guards.py
View File

@@ -14,13 +14,9 @@ from typing import Any, Dict, Optional
logger = logging.getLogger("tools.skill_manager_tool")
_ERR_KEY = "error"
def _refusal(message: str, **extra: Any) -> Dict[str, Any]:
out: Dict[str, Any] = {"success": False, _ERR_KEY: message}
out.update(extra)
return out
return {"success": False, "error": message, **extra}
def _is_background_review() -> bool:
@@ -39,9 +35,7 @@ def _resolved_str(path: Path) -> str:
return str(path)
# ---------------------------------------------------------------------------
# Background-review read marks
# ---------------------------------------------------------------------------
# --- Background-review read marks --------------------------------------------
class _BackgroundReviewReadMarks:
"""Read marks shared by copied tool contexts within one review run."""
@@ -59,18 +53,15 @@ class _BackgroundReviewReadMarks:
return path in self._paths
_background_review_read_paths: (
"_ctxvars.ContextVar[Optional[_BackgroundReviewReadMarks]]"
) = _ctxvars.ContextVar("background_review_read_paths", default=None)
_background_review_read_paths: "_ctxvars.ContextVar[Optional[_BackgroundReviewReadMarks]]" = (
_ctxvars.ContextVar("background_review_read_paths", default=None))
def mark_background_review_skill_read(path: Path) -> None:
"""Record that the active background-review fork has read a skill file.
The review fork may evolve skills but must not patch content it only
inferred from the transcript: skill_view calls this after returning file
content, and the write guards below require the target to be marked.
"""
The fork must not patch content it only inferred from the transcript:
skill_view/read_file call this, and the write guards require the mark."""
if not _is_background_review():
return
marks = _background_review_read_paths.get()
@@ -90,9 +81,7 @@ def _reset_background_review_read_marks() -> None:
_background_review_read_paths.set(_BackgroundReviewReadMarks())
# ---------------------------------------------------------------------------
# Delete-target safety
# ---------------------------------------------------------------------------
# --- Delete-target safety -----------------------------------------------------
def _containing_skills_root(skill_path: Path) -> Path:
"""Skills root (local or external_dirs entry) containing ``skill_path``;
@@ -123,20 +112,15 @@ def _is_path_redirect(path: Path) -> bool:
def _validate_delete_target(skill_dir: Path) -> Optional[str]:
"""Last-line guard before ``shutil.rmtree(skill_dir)``.
``_find_skill`` already restricts the dir to a real SKILL.md parent, but
even a poisoned tree must never recursively delete (1) a path outside every
known skills root, (2) a skills root itself, or (3) a symlink/junction
(rmtree would follow it). Returns a refusal string or ``None``.
"""
"""Last-line guard before ``shutil.rmtree(skill_dir)``: even a poisoned tree
must never delete (1) a path outside every known skills root, (2) a skills
root itself, or (3) a symlink/junction (rmtree would follow it)."""
from agent.skill_utils import get_all_skills_dirs
if _is_path_redirect(skill_dir):
return (
f"Refusing to delete '{skill_dir}': the skill directory is a "
f"symlink/junction. Remove the link target manually if intended."
)
f"symlink/junction. Remove the link target manually if intended.")
try:
resolved = skill_dir.resolve()
except OSError as exc:
@@ -150,8 +134,7 @@ def _validate_delete_target(skill_dir: Path) -> Optional[str]:
if resolved == root:
return (
f"Refusing to delete '{skill_dir}': resolves to the skills root "
f"itself, which would remove every installed skill."
)
f"itself, which would remove every installed skill.")
try:
rel = resolved.relative_to(root)
except ValueError:
@@ -159,31 +142,25 @@ def _validate_delete_target(skill_dir: Path) -> Optional[str]:
if rel.parts:
return None
return (
f"Refusing to delete '{skill_dir}': path does not resolve inside any "
f"known skills root."
f"Refusing to delete '{skill_dir}': path does not resolve inside any " f"known skills root."
)
# ---------------------------------------------------------------------------
# Ownership / provenance guards
# ---------------------------------------------------------------------------
# --- Ownership / provenance guards --------------------------------------------
def _pinned_guard(name: str) -> Optional[str]:
"""Refusal message if *name* is pinned or essential, else None.
Pin only guards against **deletion** (curator auto-archive and
``skill_manage(delete)``); patches/edits stay allowed. Essential skills
(``ESSENTIAL_SKILLS``) are permanently pinned because the system prompt
references them. Best-effort: an unreadable sidecar lets the delete through.
"""
Pin only guards **deletion**; patches/edits stay allowed. ESSENTIAL_SKILLS are
permanently pinned (the system prompt references them). Best-effort: an
unreadable sidecar lets the delete through."""
try:
from agent.skill_utils import ESSENTIAL_SKILLS
if name in ESSENTIAL_SKILLS:
return (
f"Skill '{name}' is essential to Hermes (the agent's own "
f"operating manual referenced by the system prompt) and "
f"cannot be deleted. Patches and edits are still allowed."
)
f"cannot be deleted. Patches and edits are still allowed.")
except Exception:
logger.debug("essential-guard lookup failed for %s", name, exc_info=True)
try:
@@ -194,24 +171,19 @@ def _pinned_guard(name: str) -> Optional[str]:
f"skill_manage. Ask the user to run "
f"`hermes curator unpin {name}` if they want to delete it. "
f"Patches and edits are allowed on pinned skills; only "
f"deletion is blocked."
)
f"deletion is blocked.")
except Exception:
logger.debug("pinned-guard lookup failed for %s", name, exc_info=True)
return None
def _background_review_write_guard(
name: str,
skill_dir: Path,
action: str,
name: str, skill_dir: Path, action: str,
) -> Optional[Dict[str, Any]]:
"""Refuse autonomous curator writes to anything but curator-owned sediment.
Foreground agents may edit external/bundled/hub skills at the user's
direction; the background review fork has no user in the loop, so it is
also blocked on pinned skills (stricter than ``_pinned_guard``).
"""
The background review fork has no user in the loop, so unlike foreground
agents it is also blocked on pinned/external/bundled/hub skills."""
if not _is_background_review():
return None
@@ -222,8 +194,7 @@ def _background_review_write_guard(
f"Refusing background curator {action} for pinned skill "
f"'{name}': pinned skills are off-limits to autonomous "
"maintenance. Ask the user to run "
f"`hermes curator unpin {name}` if they want it changed."
)
f"`hermes curator unpin {name}` if they want it changed.")
except Exception:
logger.debug("pinned skill guard lookup failed for %s", name, exc_info=True)
@@ -233,8 +204,7 @@ def _background_review_write_guard(
return _refusal(
f"Refusing background curator {action} for skill '{name}': "
"the skill lives in skills.external_dirs, which are "
"externally owned and read-only to autonomous curation."
)
"externally owned and read-only to autonomous curation.")
except Exception:
logger.debug("external skill guard lookup failed for %s", name, exc_info=True)
@@ -243,13 +213,10 @@ def _background_review_write_guard(
for predicate, label in (
(skill_usage.is_protected_builtin, "protected built-in"),
(skill_usage.is_hub_installed, "hub-installed"),
(skill_usage.is_bundled, "bundled"),
):
(skill_usage.is_bundled, "bundled")):
if predicate(name):
return _refusal(
f"Refusing background curator {action} for {label} "
f"skill '{name}'."
)
f"Refusing background curator {action} for {label} " f"skill '{name}'.")
# Not curator-managed (no `created_by: "agent"` marker) => user-owned.
# A MISSING record and an explicit `created_by: null` must resolve
# IDENTICALLY: keying on record presence made the policy depend on the
@@ -258,32 +225,24 @@ def _background_review_write_guard(
# both; `hermes curator adopt <name>` is the supported way in.
usage_rec = skill_usage.load_usage().get(name)
if not skill_usage._is_curator_managed_record(usage_rec):
if isinstance(usage_rec, dict):
_detail = f"created_by={usage_rec.get('created_by')!r}"
else:
_detail = "no usage record"
_detail = (f"created_by={usage_rec.get('created_by')!r}" if isinstance(usage_rec, dict)
else "no usage record")
return _refusal(
f"Refusing background curator {action} for skill "
f"'{name}': the skill is not curator-managed ({_detail}). "
"User-owned skills are off-limits to autonomous curation. "
f"Run `hermes curator adopt {name}` to opt it in."
)
f"Run `hermes curator adopt {name}` to opt it in.")
except Exception:
logger.warning("owned skill guard lookup failed for %s", name, exc_info=True)
return _refusal(
f"Refusing background curator {action} for skill '{name}': "
"agent ownership could not be verified because the provenance "
"record is unavailable or unreadable."
)
"record is unavailable or unreadable.")
return None
def _background_review_read_before_write_guard(
name: str,
target: Path,
action: str,
file_label: str,
) -> Optional[Dict[str, Any]]:
name: str, target: Path, action: str, file_label: str) -> Optional[Dict[str, Any]]:
"""Require review forks to load the exact target before mutating it."""
if not _is_background_review() or _background_review_has_read(target):
return None
@@ -293,8 +252,7 @@ def _background_review_read_before_write_guard(
"review turn. Call skill_view(name) for SKILL.md, or "
"skill_view(name, file_path=...) for a supporting file, then "
"retry the write using the content just returned.",
_read_before_write_required=True,
)
_read_before_write_required=True)
def _background_review_preflight(action: str, name: str) -> Optional[Dict[str, Any]]:
@@ -309,18 +267,14 @@ def _background_review_preflight(action: str, name: str) -> Optional[Dict[str, A
def _curator_consolidation_delete_guard(
name: str, absorbed_into: Optional[str]
name: str, absorbed_into: Optional[str],
) -> Optional[Dict[str, Any]]:
"""Fail closed on unverified deletes during the curator consolidation pass.
The review fork's only legitimate ``skill_manage(delete)`` is a verified
consolidation declared via ``absorbed_into=<umbrella>`` (existence is
validated in ``_delete_skill``). A bare delete (``None`` or ``""``) is the
fail-open behavior that once archived whole clusters of active skills; the
deterministic inactivity prune archives via ``skill_usage.archive_skill``
without ever calling ``skill_manage``, so a bare prune here can only be the
LLM pass pruning without evidence. Refuse it; keep the skill active.
"""
The review fork's only legitimate delete is a verified consolidation declared
via ``absorbed_into=<umbrella>`` (existence validated in ``_delete_skill``).
The deterministic inactivity prune never calls ``skill_manage``, so a bare
delete here can only be the LLM pass pruning without evidence: refuse it."""
if not _is_background_review():
return None
if isinstance(absorbed_into, str) and absorbed_into.strip():
@@ -333,20 +287,15 @@ def _curator_consolidation_delete_guard(
"skill with no forwarding target is not permitted here — the "
"deterministic inactivity prune handles staleness archival "
"separately. Keeping '{name}' active.".format(name=name),
_fail_closed=True,
)
_fail_closed=True)
# ---------------------------------------------------------------------------
# Org-mirror handling
# ---------------------------------------------------------------------------
# --- Org-mirror handling ------------------------------------------------------
def _maybe_auto_propose_org_edit(name: str, skill_path: Path) -> Optional[str]:
"""Submit an org-skill edit upstream when `sync.org_auto_propose` is on.
Returns a short note for the tool result, or None when nothing happened.
Never raises: the edit is already saved locally and can be proposed later.
"""
Returns a note for the tool result or None; never raises (the edit is
already saved locally and can be proposed later)."""
from tools import skill_manager_tool as _smt
try:
@@ -359,33 +308,27 @@ def _maybe_auto_propose_org_edit(name: str, skill_path: Path) -> Optional[str]:
return (
f"This skill is shared by your organisation. Your edit is "
f"saved locally and will not be overwritten by org updates. "
f"Run `hermes sync propose {name}` to share it back."
)
f"Run `hermes sync propose {name}` to share it back.")
result = ssc.propose_skill(name)
if result.get("proposal_pending"):
return (
f"Auto-proposed to your organisation as proposal "
f"#{result.get('proposal_id')} (pending admin review)."
)
f"#{result.get('proposal_id')} (pending admin review).")
return "Auto-proposed to your organisation (merged into the shared set)."
except Exception as e:
logger.debug("auto-propose skipped for %s: %s", name, e)
return (
f"Edit saved locally. Could not submit it to your organisation "
f"right now — run `hermes sync propose {name}` to retry."
)
f"right now — run `hermes sync propose {name}` to retry.")
def _org_mirror_write_guard(name: str, skill_path: Path, action: str) -> Optional[Dict[str, Any]]:
"""Org-shared skills are EDITABLE IN PLACE — this only blocks deletion.
Refusing every write to `_org/` froze shared skills while personal ones
kept improving (agents don't fork mid-task). Edits now land in the mirror,
survive the next org pull (baseline sidecar in skills_sync_client), and
reach the org via `hermes sync propose` or `sync.org_auto_propose`.
Deletion stays refused: the mirror is a view of org HEAD, so a local
delete just comes back, and removing for everyone is an admin action.
"""
Edits land in the mirror, survive the next org pull (baseline sidecar in
skills_sync_client) and reach the org via `hermes sync propose`. Deletion
stays refused: the mirror is a view of org HEAD, so a local delete just
comes back, and removing for everyone is an admin action."""
if action not in {"delete", "remove_file"}:
return None
from tools import skill_manager_tool as _smt
@@ -400,8 +343,7 @@ def _org_mirror_write_guard(name: str, skill_path: Path, action: str) -> Optiona
"the next sync. Ask an org admin to remove it for "
"everyone. (Editing it IS allowed — your changes are kept "
"and can be proposed back with `hermes sync propose "
f"{name}`.)"
)
f"{name}`.)")
except Exception:
logger.debug("org mirror guard lookup failed for %s", name, exc_info=True)
return None

277
tools/skill_manager_tool.py
View File

@@ -6,9 +6,7 @@ Lets the agent create, patch, and delete skills — its procedural memory
(narrow, actionable "how to do X"), as opposed to MEMORY.md/USER.md (broad,
declarative). New skills land in ~/.hermes/skills/ (or ``skills.create_dir``);
existing skills (bundled, hub-installed, user-created) are modified in place.
Actions: create, edit (legacy full rewrite), patch, delete, write_file,
remove_file. Layout: ``<skills>/[category/]<skill>/SKILL.md`` plus optional
Layout: ``<skills>/[category/]<skill>/SKILL.md`` plus optional
``references/ templates/ scripts/ assets/`` subdirs.
"""
@@ -30,8 +28,7 @@ from agent.skill_utils import (
extract_skill_description,
is_skill_description_truncated_for_prompt,
parse_frontmatter as _parse_frontmatter,
SKILL_PROMPT_DESC_LIMIT,
)
SKILL_PROMPT_DESC_LIMIT)
from tools.skill_manager_guards import ( # noqa: F401 — re-exported for callers/tests
_BackgroundReviewReadMarks,
_background_review_has_read,
@@ -48,24 +45,14 @@ from tools.skill_manager_guards import ( # noqa: F401 — re-exported for calle
_reset_background_review_read_marks,
_validate_delete_target,
_is_background_review,
mark_background_review_skill_read,
)
mark_background_review_skill_read)
from tools.skill_manager_batch import ( # noqa: F401
_BATCH_MAX_OPS,
_BATCH_OP_ACTIONS,
_skill_manage_batch,
_BATCH_MAX_OPS, _BATCH_OP_ACTIONS, _skill_manage_batch,
)
from tools.skills_guard import scan_skill, should_allow_install, format_scan_report
logger = logging.getLogger(__name__)
# External hub installs are always scanned; agent-created skills only when
# skills.guard_agent_created is on.
try:
from tools.skills_guard import scan_skill, should_allow_install, format_scan_report
_GUARD_AVAILABLE = True
except ImportError:
_GUARD_AVAILABLE = False
def _guard_agent_created_enabled() -> bool:
"""skills.guard_agent_created (default False): the agent can already run
@@ -73,9 +60,7 @@ def _guard_agent_created_enabled() -> bool:
try:
from hermes_cli.config import load_config
return is_truthy_value(
cfg_get(load_config(), "skills", "guard_agent_created"),
default=False,
)
cfg_get(load_config(), "skills", "guard_agent_created"), default=False)
except Exception:
return False
@@ -87,7 +72,7 @@ def _security_scan_skill(skill_dir: Path) -> Optional[str]:
verdict means dangerous findings for an agent-created skill — surfaced as
an error so the agent can retry with the flagged content removed.
"""
if not _GUARD_AVAILABLE or not _guard_agent_created_enabled():
if not _guard_agent_created_enabled():
return None
try:
result = scan_skill(skill_dir, source="agent-created")
@@ -143,9 +128,7 @@ def _err(message: str) -> Dict[str, Any]:
return {"success": False, "error": message}
# =============================================================================
# Validation helpers
# =============================================================================
# --- Validation helpers -------------------------------------------------------
def _validate_name(name: str) -> Optional[str]:
"""Validate a skill name. Returns error message or None if valid."""
@@ -154,10 +137,8 @@ def _validate_name(name: str) -> Optional[str]:
if len(name) > MAX_NAME_LENGTH:
return f"Skill name exceeds {MAX_NAME_LENGTH} characters."
if not VALID_NAME_RE.match(name):
return (
f"Invalid skill name '{name}'. Use lowercase letters, numbers, "
f"hyphens, dots, and underscores. Must start with a letter or digit."
)
return (f"Invalid skill name '{name}'. Use lowercase letters, numbers, "
f"hyphens, dots, and underscores. Must start with a letter or digit.")
return None
@@ -170,10 +151,8 @@ def _validate_category(category: Optional[str]) -> Optional[str]:
category = category.strip()
if not category:
return None
invalid = (
f"Invalid category '{category}'. Use lowercase letters, numbers, "
"hyphens, dots, and underscores. Categories must be a single directory name."
)
invalid = (f"Invalid category '{category}'. Use lowercase letters, numbers, "
"hyphens, dots, and underscores. Categories must be a single directory name.")
if "/" in category or "\\" in category:
return invalid
if len(category) > MAX_NAME_LENGTH:
@@ -215,8 +194,7 @@ def _validate_frontmatter(content: str, *, new_skill: bool = False) -> Optional[
f"{SKILL_PROMPT_DESC_LIMIT}-char system-prompt budget (one sentence, "
f"trigger first, ends with a period). The skill index truncates "
f"longer descriptions to {SKILL_PROMPT_DESC_LIMIT - 3} chars + '...', "
f"destroying the routing signal. Move detail into the skill body."
)
f"destroying the routing signal. Move detail into the skill body.")
if not content[end_match.end() + 3:].strip():
return "SKILL.md must have content after the frontmatter (instructions, procedures, etc.)."
return None
@@ -229,8 +207,7 @@ def _validate_content_size(content: str, label: str = "SKILL.md") -> Optional[st
f"{label} content is {len(content):,} characters "
f"(limit: {MAX_SKILL_CONTENT_CHARS:,}). "
f"Consider splitting into a smaller SKILL.md with supporting files "
f"in references/ or templates/."
)
f"in references/ or templates/.")
return None
@@ -288,8 +265,7 @@ def _find_skill(name: str) -> Optional[Dict[str, Any]]:
except OSError:
logger.debug(
"skills dir resolve failed; categorized lookups fall back to the unresolved path",
exc_info=True,
)
exc_info=True)
local_root = _skills_dir()
for skills_dir in get_all_skills_dirs():
@@ -309,11 +285,8 @@ def _find_skill(name: str) -> Optional[Dict[str, Any]]:
def _find_skill_in_other_profiles(name: str) -> List[Tuple[str, Path]]:
"""``(profile_name, skill_dir)`` pairs for OTHER profiles holding ``name``.
Lets the "not found" error explain a wrong-profile mistake. Fail-quiet:
empty list when discovery fails.
"""
"""``(profile_name, skill_dir)`` pairs for OTHER profiles holding ``name``;
lets the "not found" error explain a wrong-profile mistake. Fail-quiet."""
matches: List[Tuple[str, Path]] = []
try:
from hermes_constants import get_default_hermes_root
@@ -360,15 +333,13 @@ def _skill_not_found_error(name: str, suffix: str = "") -> str:
f" A skill by that name exists in profile "
f"'{other_profile}' ({other_path}). To edit it, switch "
f"profiles (`hermes -p {other_profile}`) or edit the file "
f"directly (file tools / terminal)."
)
f"directly (file tools / terminal).")
elif others:
names = ", ".join(f"'{p}'" for p, _ in others)
base += (
f" Skills by that name exist in other profiles: {names}. "
f"Switch profiles (`hermes -p <name>`) to edit there, or "
f"edit the files directly (file tools / terminal)."
)
f"edit the files directly (file tools / terminal).")
else:
base += " Use skills_list() to see available skills."
return base + suffix
@@ -396,21 +367,16 @@ def _validate_file_path(file_path: str) -> Optional[str]:
return None
def _resolve_skill_target(skill_dir: Path, file_path: str) -> Tuple[Optional[Path], Optional[str]]:
"""Resolve a supporting-file path and ensure it stays within the skill directory."""
def _resolve_supporting_file(skill_dir: Path, file_path: str):
"""Validate ``file_path`` and resolve it inside ``skill_dir``.
Returns (target, None) or (None, error_dict)."""
from tools.path_security import validate_within_dir
target = skill_dir / file_path
error = validate_within_dir(target, skill_dir)
return (None, error) if error else (target, None)
def _resolve_supporting_file(skill_dir: Path, file_path: str):
"""``_validate_file_path`` + ``_resolve_skill_target``. Returns (target, None) or (None, error_dict)."""
err = _validate_file_path(file_path)
if err:
return None, _err(err)
target, err = _resolve_skill_target(skill_dir, file_path)
target = skill_dir / file_path
err = validate_within_dir(target, skill_dir)
if err:
return None, _err(err)
return target, None
@@ -418,17 +384,13 @@ def _resolve_supporting_file(skill_dir: Path, file_path: str):
def _locate_for_write(name: str, action: str, not_found_suffix: str = ""):
"""Find the skill and run the org-mirror + background-review write guards.
Returns ``(skill_dir, None)`` or ``(None, error_dict)``.
"""
Returns ``(skill_dir, None)`` or ``(None, error_dict)``."""
existing = _find_skill(name)
if not existing:
return None, _err(_skill_not_found_error(name, not_found_suffix))
skill_dir = existing["path"]
guard = (
_org_mirror_write_guard(name, skill_dir, action)
or _background_review_write_guard(name, skill_dir, action)
)
guard = (_org_mirror_write_guard(name, skill_dir, action)
or _background_review_write_guard(name, skill_dir, action))
if guard:
return None, guard
return skill_dir, None
@@ -455,11 +417,6 @@ def _attach_org_note(result: Dict[str, Any], name: str, skill_dir: Path) -> None
result["message"] = f"{result['message']} {org_note}"
# =============================================================================
# Core actions
# =============================================================================
def _add_description_prompt_preview(result: Dict[str, Any], content: str) -> None:
"""Append a system_prompt_preview field when the description will be truncated."""
fm, _ = _parse_frontmatter(content)
@@ -467,8 +424,7 @@ def _add_description_prompt_preview(result: Dict[str, Any], content: str) -> Non
result["system_prompt_preview"] = (
f"System prompt will show: \"{extract_skill_description(fm)}\" — "
f"keep the trigger self-contained in the first "
f"{SKILL_PROMPT_DESC_LIMIT - 3} chars."
)
f"{SKILL_PROMPT_DESC_LIMIT - 3} chars.")
def _attach_lint_findings(result: Dict[str, Any], skill_md: Path) -> None:
@@ -476,31 +432,28 @@ def _attach_lint_findings(result: Dict[str, Any], skill_md: Path) -> None:
hard rejects already ran in _validate_frontmatter)."""
try:
from tools.skill_linter import lint_skill # local import: optional path
findings = lint_skill(skill_md)
except Exception:
return
if not findings:
return
result["lint_warnings"] = [
{"severity": f.severity, "rule": f.rule, "message": f.message}
for f in findings
]
{"severity": f.severity, "rule": f.rule, "message": f.message} for f in findings]
result["lint_hint"] = (
"The skill was created. These are advisory authoring-convention "
"findings (not blockers) — fix them with skill_manage(action='patch') "
"to match Hermes skill standards."
)
"to match Hermes skill standards.")
# --- Core actions -------------------------------------------------------------
def _create_skill(name: str, content: str, category: str = None) -> Dict[str, Any]:
"""Create a new user skill with SKILL.md content."""
err = (
_validate_name(name)
or _validate_category(category)
or _validate_frontmatter(content, new_skill=True)
or _validate_content_size(content)
)
or _validate_content_size(content))
if err:
return _err(err)
existing = _find_skill(name)
@@ -525,8 +478,7 @@ def _create_skill(name: str, content: str, category: str = None) -> Dict[str, An
"message": f"Skill '{name}' created.",
"path": _display_path,
"skill_md": str(skill_md),
"_change": {"description": _description_preview(content)},
}
"_change": {"description": _description_preview(content)}}
if category:
result["category"] = category
result["hint"] = (
@@ -561,20 +513,14 @@ def _edit_skill(name: str, content: str) -> Dict[str, Any]:
"success": True,
"message": f"Skill '{name}' updated (full rewrite).",
"path": str(skill_dir),
"_change": {"description": _description_preview(content)},
}
"_change": {"description": _description_preview(content)}}
_attach_org_note(result, name, skill_dir)
_add_description_prompt_preview(result, content)
return result
def _patch_skill(
name: str,
old_string: str,
new_string: str,
file_path: str = None,
replace_all: bool = False,
) -> Dict[str, Any]:
def _patch_skill(name: str, old_string: str, new_string: str, file_path: str = None,
replace_all: bool = False) -> Dict[str, Any]:
"""Targeted find-and-replace within SKILL.md (default) or a supporting file.
Requires a unique match unless replace_all is True."""
if not old_string:
@@ -585,8 +531,7 @@ def _patch_skill(
"file. Read the target file first (read_file on the skill's SKILL.md, or the file "
"named by file_path) and copy the snippet verbatim, then retry 'patch'. "
"Do NOT fall back to action='write_file' — that rewrites the entire file and "
"destroys unrelated content."
)
"destroys unrelated content.")
if new_string is None:
return _err("new_string is required for 'patch'. Use an empty string to delete matched text.")
# No old_string == new_string guard here: fuzzy_find_and_replace rejects
@@ -616,8 +561,7 @@ def _patch_skill(
from tools.fuzzy_match import fuzzy_find_and_replace
new_content, match_count, _strategy, match_error = fuzzy_find_and_replace(
content, old_string, new_string, replace_all
)
content, old_string, new_string, replace_all)
if match_error:
try:
from tools.fuzzy_match import format_no_match_hint
@@ -643,9 +587,7 @@ def _patch_skill(
"message": f"Patched {target_label} in skill '{name}' ({match_count} replacement{'s' if match_count > 1 else ''}).",
"_change": {
"old": old_string[:200] + ("…" if len(old_string) > 200 else ""),
"new": new_string[:200] + ("…" if len(new_string) > 200 else ""),
},
}
"new": new_string[:200] + ("…" if len(new_string) > 200 else "")}}
_attach_org_note(result, name, skill_dir)
return result
@@ -676,8 +618,7 @@ def _delete_skill(name: str, absorbed_into: Optional[str] = None) -> Dict[str, A
if not _find_skill(absorbed_target):
return _err(
f"absorbed_into='{absorbed_target}' does not exist. "
f"Create or patch the umbrella skill first, then retry the delete."
)
f"Create or patch the umbrella skill first, then retry the delete.")
skills_root = _containing_skills_root(skill_dir)
unsafe = _validate_delete_target(skill_dir) # defense-in-depth before rmtree
@@ -699,8 +640,7 @@ def _delete_skill(name: str, absorbed_into: Optional[str] = None) -> Dict[str, A
return {
"success": True,
"message": f"Skill '{name}' archived ({archive_msg}).{absorbed_note}",
"_archived": True,
}
"_archived": True}
shutil.rmtree(skill_dir)
_rmdir_if_empty(skill_dir.parent, skills_root) # empty category dir, never the root
@@ -724,8 +664,7 @@ def _write_file(name: str, file_path: str, file_content: str) -> Dict[str, Any]:
return _err(
f"File content is {content_bytes:,} bytes "
f"(limit: {MAX_SKILL_FILE_BYTES:,} bytes / 1 MiB). "
f"Consider splitting into smaller files."
)
f"Consider splitting into smaller files.")
err = _validate_content_size(file_content, label=file_path)
if err:
return _err(err)
@@ -749,8 +688,7 @@ def _write_file(name: str, file_path: str, file_content: str) -> Dict[str, Any]:
result = {
"success": True,
"message": f"File '{file_path}' written to skill '{name}'.",
"path": str(target),
}
"path": str(target)}
_attach_org_note(result, name, skill_dir)
return result
@@ -777,13 +715,11 @@ def _remove_file(name: str, file_path: str) -> Dict[str, Any]:
for subdir in ALLOWED_SUBDIRS
if (skill_dir / subdir).exists()
for f in (skill_dir / subdir).rglob("*")
if f.is_file()
]
if f.is_file()]
return {
"success": False,
"error": f"File '{file_path}' not found in skill '{name}'.",
"available_files": available if available else None,
}
"available_files": available if available else None}
read_guard = _background_review_read_before_write_guard(name, target, "remove_file", file_path)
if read_guard:
return read_guard
@@ -793,15 +729,12 @@ def _remove_file(name: str, file_path: str) -> Dict[str, Any]:
return {"success": True, "message": f"File '{file_path}' removed from skill '{name}'."}
# =============================================================================
# Main entry point
# =============================================================================
# --- Main entry point ---------------------------------------------------------
# Set while replaying an already-approved staged skill write so skill_manage()
# does not re-gate (and re-stage) it.
_skill_gate_bypass: "_ctxvars.ContextVar[bool]" = _ctxvars.ContextVar(
"skill_gate_bypass", default=False
)
"skill_gate_bypass", default=False)
_GATED_ACTIONS = {"create", "edit", "patch", "delete", "write_file", "remove_file"}
@@ -825,8 +758,7 @@ def _run_write_gate(build_staging):
return json.dumps(
{"success": True, "staged": True, "pending_id": record["id"],
"gist": gist, "message": decision.message},
ensure_ascii=False,
)
ensure_ascii=False)
def _apply_skill_write_gate(action, name, **payload_kwargs):
@@ -843,15 +775,13 @@ def _apply_skill_write_gate(action, name, **payload_kwargs):
content=payload_kwargs.get("content") or "",
file_path=payload_kwargs.get("file_path") or "",
old_string=payload_kwargs.get("old_string") or "",
new_string=payload_kwargs.get("new_string") or "",
)
new_string=payload_kwargs.get("new_string") or "")
return payload, gist
return _run_write_gate(_staging)
_FLAT_OP_KEYS = ("content", "category", "file_path", "file_content",
"old_string", "new_string")
_FLAT_OP_KEYS = ("content", "category", "file_path", "file_content", "old_string", "new_string")
def _skill_manage_from(payload: Dict[str, Any], **extra) -> str:
@@ -861,8 +791,7 @@ def _skill_manage_from(payload: Dict[str, Any], **extra) -> str:
name=payload.get("name", ""),
replace_all=payload.get("replace_all", False),
**{k: payload.get(k) for k in _FLAT_OP_KEYS},
**extra,
)
**extra)
def apply_skill_pending(payload: Dict[str, Any]) -> str:
@@ -871,10 +800,8 @@ def apply_skill_pending(payload: Dict[str, Any]) -> str:
token = _skill_gate_bypass.set(True)
try:
return _skill_manage_from(
payload,
absorbed_into=payload.get("absorbed_into"),
operations=payload.get("operations"),
)
payload, absorbed_into=payload.get("absorbed_into"),
operations=payload.get("operations"))
finally:
_skill_gate_bypass.reset(token)
@@ -891,12 +818,11 @@ def _maybe_debounced_sync_push(skill_name: str) -> None:
Fast-path: skills not opted into sync do nothing (no auth, no network).
The push runs via ``skills_sync_client.maybe_push_skills`` which enforces
the access gate and swallows errors. Never blocks the caller (M1-C).
the access gate and swallows errors. Never blocks the caller.
"""
global _sync_push_timer, _sync_push_lock
try:
from tools.skill_usage import is_sync_enabled
if not is_sync_enabled(skill_name):
return
except Exception:
@@ -908,7 +834,6 @@ def _maybe_debounced_sync_push(skill_name: str) -> None:
def _fire():
try:
from tools.skills_sync_client import maybe_push_skills
maybe_push_skills(message=f"sync: {skill_name}")
except Exception:
pass
@@ -941,8 +866,7 @@ def _act_patch(a):
return tool_error(
"Pass EITHER content (full SKILL.md rewrite) OR "
"old_string/new_string (targeted replacement), not both.",
success=False,
)
success=False)
if a["content"]:
return _edit_skill(a["name"], a["content"])
# Targeted-replacement validation lives in _patch_skill so the public
@@ -972,8 +896,7 @@ _ACTION_HANDLERS = {
"patch": _act_patch,
"delete": lambda a: _delete_skill(a["name"], absorbed_into=a["absorbed_into"]),
"write_file": _act_write_file,
"remove_file": _act_remove_file,
}
"remove_file": _act_remove_file}
def _record_success(action, name, result, *, file_path, absorbed_into, task_id,
@@ -993,12 +916,10 @@ def _record_success(action, name, result, *, file_path, absorbed_into, task_id,
if file_path:
_evidence["file_path"] = file_path
_ledger.record_mutation(
action,
name,
action, name,
before=ledger_before if ledger_before is not None else [],
after_root=_post["path"] if _post else None,
evidence=_evidence,
)
evidence=_evidence)
except Exception:
pass
try:
@@ -1032,20 +953,10 @@ def _record_success(action, name, result, *, file_path, absorbed_into, task_id,
def skill_manage(
action: str,
name: str,
content: str = None,
category: str = None,
file_path: str = None,
file_content: str = None,
old_string: str = None,
new_string: str = None,
replace_all: bool = False,
absorbed_into: str = None,
task_id: str = None,
session_id: str = None,
operations=None,
) -> str:
action: str, name: str, content: str = None, category: str = None, file_path: str = None,
file_content: str = None, old_string: str = None, new_string: str = None,
replace_all: bool = False, absorbed_into: str = None, task_id: str = None,
session_id: str = None, operations=None) -> str:
"""Manage user-created skills; dispatches to the action handler.
``operations``: batch shape — a list of {action, ...} dicts applied
@@ -1054,9 +965,7 @@ def skill_manage(
"""
if operations is not None:
return _skill_manage_batch(
operations, default_name=name or None,
task_id=task_id, session_id=session_id,
)
operations, default_name=name or None, task_id=task_id, session_id=session_id)
preflight = _background_review_preflight(action, name)
if preflight is not None:
return json.dumps(preflight, ensure_ascii=False)
@@ -1067,8 +976,7 @@ def skill_manage(
args = dict(
content=content, category=category, file_path=file_path,
file_content=file_content, old_string=old_string, new_string=new_string,
replace_all=replace_all, absorbed_into=absorbed_into,
)
replace_all=replace_all, absorbed_into=absorbed_into)
gate_result = _apply_skill_write_gate(action, name, **args)
if gate_result is not None:
return gate_result
@@ -1082,10 +990,7 @@ def skill_manage(
from tools import skill_ledger as _ledger
_pre = _find_skill(name)
_ledger_before = _ledger.capture_before(
_pre["path"] if _pre else None,
complete_package=(action == "delete"),
skill=name,
)
_pre["path"] if _pre else None, complete_package=(action == "delete"), skill=name)
except Exception:
pass
@@ -1100,14 +1005,11 @@ def skill_manage(
if result.get("success"):
_record_success(
action, name, result, file_path=file_path, absorbed_into=absorbed_into,
task_id=task_id, session_id=session_id, ledger_before=_ledger_before,
)
task_id=task_id, session_id=session_id, ledger_before=_ledger_before)
return json.dumps(result, ensure_ascii=False)
# =============================================================================
# OpenAI Function-Calling Schema
# =============================================================================
# --- OpenAI Function-Calling Schema -------------------------------------------
SKILL_MANAGE_SCHEMA = {
"name": "skill_manage",
@@ -1128,8 +1030,7 @@ SKILL_MANAGE_SCHEMA = {
"op only). Existing skills are modified wherever they live. Keep "
"the description's first 57 chars a self-contained trigger: 'Use "
"when <trigger>. <one-line behavior>.' — skill_view() shows "
"format conventions."
),
"format conventions."),
"parameters": {
"type": "object",
"properties": {
@@ -1144,25 +1045,19 @@ SKILL_MANAGE_SCHEMA = {
"description": (
"Skill name (lowercase, hyphens/underscores, "
"max 64 chars); an existing skill's name "
"unless creating."
)
},
"unless creating.")},
"action": {
"type": "string",
"enum": ["create", "patch", "delete", "write_file", "remove_file"]
},
"enum": ["create", "patch", "delete", "write_file", "remove_file"]},
"content": {
"type": "string",
"description": (
"Full SKILL.md text (YAML frontmatter + "
"markdown body) for create, or a full "
"rewrite on patch."
)
},
"rewrite on patch.")},
"category": {
"type": "string",
"description": "Optional category subdir for create (e.g. 'devops')."
},
"description": "Optional category subdir for create (e.g. 'devops')."},
# patch args: same fuzzy-matching semantics as the
# `patch` tool — teach only skill-specific facts here.
"old_string": {
@@ -1171,12 +1066,10 @@ SKILL_MANAGE_SCHEMA = {
},
"new_string": {
"type": "string",
"description": "Replacement (patch); empty string deletes the match."
},
"description": "Replacement (patch); empty string deletes the match."},
"replace_all": {
"type": "boolean",
"description": "patch: replace all occurrences (default false)."
},
"description": "patch: replace all occurrences (default false)."},
"file_path": {
"type": "string",
"description": (
@@ -1185,17 +1078,10 @@ SKILL_MANAGE_SCHEMA = {
"never absolute. write_file/remove_file: "
"required; first segment references/, "
"templates/, scripts/, or assets/. patch: "
"optional (default SKILL.md)."
)
},
"optional (default SKILL.md).")},
"file_content": {
"type": "string",
"description": "Content for write_file."
}
},
"required": ["name", "action"]
}
},
"type": "string", "description": "Content for write_file."}},
"required": ["name", "action"]}},
# NOTE: the handler also accepts the legacy flat single-op shape
# (top-level action/name/content/old_string/new_string/
# replace_all/category/file_path/file_content) — old transcripts
@@ -1204,9 +1090,7 @@ SKILL_MANAGE_SCHEMA = {
# documents it and the delete guard's error re-teaches it).
# None are advertised.
},
"required": ["operations"],
},
}
"required": ["operations"]}}
# --- Registry ---
@@ -1222,5 +1106,4 @@ registry.register(
operations=args.get("operations"),
task_id=kw.get("task_id"),
session_id=kw.get("session_id")),
emoji="📝",
)
emoji="📝")