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

refactor(tools): skill_manage — fold gate/flat-op plumbing, shared root/pin/org helpers in guards, ledger tarball lookup inline; compact docstrings

Browse Source
This commit is contained in:
Teknium
2026-09-03 01:17:57 -07:00
parent d9d25c89ba
commit eeb86e4b6f
4 changed files with 124 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
tools/skill_ledger.py
View File

@@ -126,11 +126,10 @@ def read_blob(sha256: str) -> Optional[bytes]:
def snapshot_paths(root: Optional[Path], *, complete_package: bool = False) -> List[Dict[str, str]]:
"""{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`` unions in the newest curator tarball's files (disk hashes win)."""
"""{path, sha256} for every file under *root*, each stored as a blob; [] 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)
@@ -174,34 +173,24 @@ def package_prefixes(
candidates = [_package_rel(Path(root)) if root is not None else None]
candidates += [_package_rel(p) for p in _skill_md_parents(before)]
candidates += [skill, _strip_archive_timestamp(skill) if skill else None]
found: List[str] = []
for prefix in candidates:
prefix = (prefix or "").strip("/")
if prefix and prefix not in found:
found.append(prefix)
return found
def _latest_skills_tarball() -> Optional[Path]:
"""Newest ``skills.tar.gz`` under ``skills/.curator_backups/``."""
backups = _skills_dir() / ".curator_backups"
try:
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()]
# Parent dirs sort lexicographically == chronologically for the id shape.
return max(candidates, key=lambda p: p.parent.name) if candidates else None
return list(dict.fromkeys(p for p in ((c or "").strip("/") for c in candidates) if p))
def _read_package_files_from_latest_backup(prefixes: List[str]) -> Dict[str, bytes]:
"""``{posix-relpath: bytes}`` under *prefixes* in the newest snapshot; malicious
member names (absolute, ``..`` traversal) are rejected."""
archive = _latest_skills_tarball() if prefixes else None
if archive is None:
"""``{posix-relpath: bytes}`` under *prefixes* in the newest ``skills/.curator_backups/*/
skills.tar.gz``; malicious member names (absolute, ``..`` traversal) are rejected."""
backups = _skills_dir() / ".curator_backups"
try:
children = list(backups.iterdir()) if prefixes and backups.is_dir() else []
except OSError:
return {}
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 {}
# Parent dirs sort lexicographically == chronologically for the id shape.
archive = max(candidates, key=lambda p: p.parent.name)
prefixed = tuple(p if p.endswith("/") else p + "/" for p in prefixes)
exact = set(prefixes)
out: Dict[str, bytes] = {}
@@ -225,14 +214,12 @@ 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]]:
"""Union missing skill-package files from the newest curator snapshot.
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."""
"""Union missing skill-package files from the newest curator snapshot. 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:
@@ -247,9 +234,7 @@ def fill_snapshot_from_curator_backup(
skills = _skills_dir()
dest_root = Path(root) if root is not None else None
pkg_names = {dest_root.name, _strip_archive_timestamp(dest_root.name)} if dest_root else set()
have = {
rel for rel in (_rel_posix(str(item.get("path", "")), skills) for item in out) if rel is not None
}
have = {rel for rel in (_rel_posix(str(i.get("path", "")), skills) for i in out) if rel is not None}
for rel, data in extra.items():
parts = rel.split("/")
if dest_root is not None and parts and parts[0] in pkg_names:
@@ -298,10 +283,9 @@ 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]:
"""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."""
"""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:
@@ -375,9 +359,8 @@ def _validate_entry_paths(entry: Dict[str, Any]) -> Optional[str]:
def rollback_entry(entry_id: str) -> Tuple[bool, str]:
"""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."""
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."""
entry = get_entry(entry_id)
if entry is None:
return False, f"no ledger entry with id '{entry_id}'"

2
tools/skill_manager_batch.py
View File

@@ -169,7 +169,7 @@ def _skill_manage_batch(operations, default_name: str = None, task_id: str = Non
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], "operations": None}, task_id=task_id, session_id=session_id)
try:
parsed = json.loads(raw)
except Exception: # noqa: BLE001

67
tools/skill_manager_guards.py
View File

@@ -1,9 +1,7 @@
"""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 keep working.
"""
"""Write/delete guards for ``skill_manage``. Every guard returns ``None`` when the
operation may proceed, else a refusal (error dict or message). Origin-owned state
(``_find_skill``, ``_skills_dir``) is reached lazily via ``tools.skill_manager_tool``
so test patches keep working."""
import contextvars as _ctxvars
import logging
@@ -56,10 +54,9 @@ _background_review_read_paths: "_ctxvars.ContextVar[Optional[_BackgroundReviewRe
def mark_background_review_skill_read(path: Path) -> None:
"""Record that the active background-review fork has read a skill file.
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."""
"""Record that the active background-review fork has read a skill file. 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()
@@ -79,8 +76,7 @@ def _reset_background_review_read_marks() -> None:
def _resolved_roots(skill_path: Path):
"""``(resolved skill_path, [(root, resolved_root), ...])`` over every skills root
whose resolve() succeeds; an unresolvable skill_path is used as-is."""
"""``(resolved skill_path, [(root, resolved_root), ...])`` over every resolvable skills root."""
from agent.skill_utils import get_all_skills_dirs
try:
@@ -103,8 +99,7 @@ def _containing_skills_root(skill_path: Path) -> Path:
def _is_path_redirect(path: Path) -> bool:
"""Symlink or (Windows 3.12+) 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 rmtree outside."""
try:
return path.is_symlink() or (hasattr(path, "is_junction") and path.is_junction())
except OSError:
@@ -112,9 +107,8 @@ def _is_path_redirect(path: Path) -> bool:
def _validate_delete_target(skill_dir: Path) -> Optional[str]:
"""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)."""
"""Last-line guard before rmtree: even a poisoned tree must never delete (1) a path outside
every known skills root, (2) a skills root itself, (3) a symlink/junction (rmtree follows it)."""
if _is_path_redirect(skill_dir):
return (
f"Refusing to delete '{skill_dir}': the skill directory is a "
@@ -147,11 +141,9 @@ def _is_pinned(name: str, what: str) -> Optional[bool]:
def _pinned_guard(name: str) -> Optional[str]:
"""Refusal message if *name* is pinned or essential, else None.
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."""
"""Refusal message if *name* is pinned or essential, else None. 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:
@@ -171,10 +163,8 @@ def _pinned_guard(name: str) -> Optional[str]:
def _background_review_write_guard(
name: str, skill_dir: Path, action: str) -> Optional[Dict[str, Any]]:
"""Refuse autonomous curator writes to anything but curator-owned sediment.
The background review fork has no user in the loop, so unlike foreground
agents it is also blocked on pinned/external/bundled/hub skills."""
"""Refuse autonomous curator writes to anything but curator-owned sediment. The review fork
has no user in the loop, so it is also blocked on pinned/external/bundled/hub skills."""
if not _is_background_review():
return None
refuse = f"Refusing background curator {action} for"
@@ -244,12 +234,10 @@ def _background_review_preflight(action: str, name: str) -> Optional[Dict[str, A
def _curator_consolidation_delete_guard(
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 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."""
"""Fail closed on unverified deletes during the curator consolidation pass. The fork's only
legitimate delete is a 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."""
if not _is_background_review() or (isinstance(absorbed_into, str) and absorbed_into.strip()):
return None
return _refusal(
@@ -268,9 +256,8 @@ def _is_org_mirror(skill_path: Path) -> bool:
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 note for the tool result or None; never raises (the edit is
already saved locally and can be proposed later)."""
"""Submit an org-skill edit upstream when `sync.org_auto_propose` is on. Returns a note for
the tool result or None; never raises (the edit is saved locally and can be proposed later)."""
try:
from tools import skills_sync_client as ssc
@@ -295,12 +282,10 @@ def _maybe_auto_propose_org_edit(name: str, skill_path: Path) -> Optional[str]:
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.
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."""
"""Org-shared skills are EDITABLE IN PLACE — this only blocks deletion. 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
try:

163
tools/skill_manager_tool.py
View File

@@ -1,11 +1,10 @@
#!/usr/bin/env python3
"""Skill Manager Tool — agent-managed skill creation & editing.
Skills are the agent's procedural memory (narrow "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, user) are modified in
place. Layout: ``<skills>/[category/]<skill>/SKILL.md`` + optional
``references/ templates/ scripts/ assets/``.
Skills are the agent's procedural memory (narrow "how to do X"; MEMORY.md/USER.md are
broad, declarative). New skills land in ~/.hermes/skills/ (or ``skills.create_dir``);
existing skills (bundled, hub, user) are modified in place. Layout:
``<skills>/[category/]<skill>/SKILL.md`` + optional ``references/ templates/ scripts/ assets/``.
"""
import contextvars as _ctxvars
@@ -43,8 +42,7 @@ logger = logging.getLogger(__name__)
def _guard_agent_created_enabled() -> bool:
"""skills.guard_agent_created (default False): the agent can already run the same
code via terminal() ungated, so the scan is opt-in belt-and-suspenders."""
"""skills.guard_agent_created (default False): opt-in — terminal() runs the same code ungated."""
try:
from hermes_cli.config import load_config
return is_truthy_value(
@@ -54,9 +52,8 @@ def _guard_agent_created_enabled() -> bool:
def _security_scan_skill(skill_dir: Path) -> Optional[str]:
"""Post-write scan; error string if blocked, else None. No-op unless
skills.guard_agent_created. An "ask" verdict (dangerous findings) is surfaced
as an error so the agent can retry with the flagged content removed."""
"""Post-write scan (opt-in); error string if blocked, else None. An "ask" verdict
(dangerous findings) is surfaced as an error so the agent can retry without them."""
if not _guard_agent_created_enabled():
return None
try:
@@ -78,9 +75,8 @@ _SKILLS_DIR_AT_IMPORT = SKILLS_DIR
def _skills_dir() -> Path:
"""Active profile's skills dir at call time: multi-profile runtimes import once
and bind a different profile per session. An explicitly patched module-level
``SKILLS_DIR`` (tests) wins, otherwise resolve from the live HERMES_HOME."""
"""Active profile's skills dir at call time (multi-profile runtimes rebind per session).
An explicitly patched module-level ``SKILLS_DIR`` (tests) wins over the live HERMES_HOME."""
configured = Path(SKILLS_DIR)
return configured if configured != _SKILLS_DIR_AT_IMPORT else get_hermes_home() / "skills"
@@ -134,11 +130,9 @@ def _validate_category(category: Optional[str]) -> Optional[str]:
def _validate_frontmatter(content: str, *, new_skill: bool = False) -> Optional[str]:
"""Validate frontmatter (name + description) and a non-empty body.
``new_skill`` (create only) also enforces SKILL_PROMPT_DESC_LIMIT so new skills
never lose routing signal to index truncation; edit/patch skip it so existing
over-limit skills remain maintainable."""
"""Validate frontmatter (name + description) and a non-empty body. ``new_skill`` (create
only) also enforces SKILL_PROMPT_DESC_LIMIT so new skills never lose routing signal to
index truncation; edit/patch skip it so existing over-limit skills stay maintainable."""
if not content.strip():
return "Content cannot be empty."
content = content.lstrip("\ufeff") # tolerate a Windows UTF-8 BOM
@@ -197,7 +191,7 @@ def _resolve_skill_dir(name: str, category: str = None) -> Path:
base = get_skill_create_dir() or base
except Exception:
logger.debug("skills.create_dir lookup failed", exc_info=True)
return base / category / name if category else base / name
return base / (category or "") / name
def _iter_skill_dirs(root: Path):
@@ -208,16 +202,13 @@ def _iter_skill_dirs(root: Path):
def _find_skill(name: str) -> Optional[Dict[str, Any]]:
"""Find a skill across the local skills dir then skills.external_dirs.
"""Find a skill (local skills dir, then skills.external_dirs) -> ``{"path": Path}`` | None.
Accepts the bare dir name (``axolotl``) and the categorized relative path
(``mlops/axolotl``) — the two forms skill_view resolves. Bare lookups compare
the skill's own dir name so category-nested skills still match.
Returns ``{"path": Path}`` or None."""
Accepts the bare dir name (``axolotl``; matches category-nested skills too) and the
categorized relative path (``mlops/axolotl``) — the two forms skill_view resolves. The
categorized form matches RELATIVE to the local root only (relative_to raises for external dirs)."""
from agent.skill_utils import get_all_skills_dirs
# The categorized form matches RELATIVE to the local root only (relative_to
# raises for external dirs).
local_root = None
if "/" in name or "\\" in name:
try:
@@ -243,8 +234,8 @@ def _find_skill(name: str) -> Optional[Dict[str, Any]]:
def _find_skill_in_other_profiles(name: str) -> List[Tuple[str, Path]]:
"""``(profile, skill_dir)`` pairs for OTHER profiles holding ``name`` (so the
not-found error can explain a wrong-profile mistake). Fail-quiet."""
"""``(profile, skill_dir)`` pairs for OTHER profiles holding ``name`` (so the not-found
error can explain a wrong-profile mistake). Fail-quiet."""
matches: List[Tuple[str, Path]] = []
try:
from hermes_constants import get_default_hermes_root
@@ -255,10 +246,9 @@ def _find_skill_in_other_profiles(name: str) -> List[Tuple[str, Path]]:
active_dir = _active.resolve() if _active.exists() else _active
# Every profile's skills dir EXCEPT the active one (already searched).
candidates: List[Tuple[str, Path]] = [("default", root / "skills")]
profiles_root = root / "profiles"
with suppress(OSError):
if profiles_root.is_dir():
candidates += [(e.name, e / "skills") for e in profiles_root.iterdir() if e.is_dir()]
if (root / "profiles").is_dir():
candidates += [(e.name, e / "skills") for e in (root / "profiles").iterdir() if e.is_dir()]
for profile_name, skills_dir in candidates:
with suppress(OSError, RuntimeError):
if skills_dir.resolve() == active_dir or not skills_dir.is_dir():
@@ -323,8 +313,8 @@ def _resolve_supporting_file(skill_dir: Path, file_path: str):
def _locate_for_write(name: str, action: str, not_found_suffix: str = "", *,
org_guard: bool = True):
"""Find the skill and run the org-mirror (unless ``org_guard=False``) +
background-review write guards -> ``(skill_dir, None)`` | ``(None, error_dict)``."""
"""Find the skill; run the org-mirror (unless ``org_guard=False``) and background-review
write guards -> ``(skill_dir, None)`` | ``(None, error_dict)``."""
existing = _find_skill(name)
if not existing:
return None, _err(_skill_not_found_error(name, not_found_suffix))
@@ -336,9 +326,8 @@ def _locate_for_write(name: str, action: str, not_found_suffix: str = "", *,
def _guarded_write(name: str, skill_dir: Path, target: Path, action: str, label: str,
content: str) -> Optional[Dict[str, Any]]:
"""Read-before-write guard (existing targets only), atomic write, then the
security scan; a blocked scan restores the original (or unlinks a new file).
Returns an error dict or None."""
"""Read-before-write guard (existing targets only), atomic write, then the security scan;
a blocked scan restores the original (or unlinks a new file). Error dict or None."""
original = None
if target.exists():
if read_guard := _background_review_read_before_write_guard(name, target, action, label):
@@ -413,13 +402,11 @@ def _create_skill(name: str, content: str, category: str = None) -> Dict[str, An
display = skill_dir.relative_to(root) if skill_dir.is_relative_to(root) else skill_dir
result = {
"success": True, "message": f"Skill '{name}' created.", "path": str(display),
"skill_md": str(skill_md), "_change": {"description": _description_preview(content)}}
if category:
result["category"] = category
result["hint"] = (
"To add reference files, templates, or scripts, use "
"skill_manage(action='write_file', name='{}', file_path='references/example.md', file_content='...')".format(name)
)
"skill_md": str(skill_md), "_change": {"description": _description_preview(content)},
**({"category": category} if category else {}),
"hint": "To add reference files, templates, or scripts, use "
f"skill_manage(action='write_file', name='{name}', file_path='references/example.md', "
"file_content='...')"}
_add_description_prompt_preview(result, content)
_attach_lint_findings(result, skill_md)
return result
@@ -444,11 +431,10 @@ def _edit_skill(name: str, content: str) -> 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."""
"""Targeted find-and-replace in SKILL.md (default) or a supporting file; unique match unless replace_all."""
if not old_string:
# A bare "required" error is a dead end: the model retries blindly and
# often escapes to action='write_file', clobbering the whole file.
# A bare "required" error is a dead end: the model retries blindly and often
# escapes to action='write_file', clobbering the whole file.
return _err(
"old_string is required for 'patch' and must be the EXACT text currently in the file. "
"Read the target file first (read_file on the skill's SKILL.md, or the file named by "
@@ -456,12 +442,12 @@ def _patch_skill(name: str, old_string: str, new_string: str, file_path: str = N
"action='write_file' — that rewrites the entire file and 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
# that with a richer error (file_preview) this layer cannot produce.
# No old_string == new_string guard here: fuzzy_find_and_replace rejects that with a
# richer error (file_preview) this layer cannot produce.
skill_dir, guard = _locate_for_write(name, "patch")
if guard:
return guard
target_label = file_path or "SKILL.md"
if file_path:
target, err = _resolve_supporting_file(skill_dir, file_path)
if err:
@@ -470,13 +456,12 @@ def _patch_skill(name: str, old_string: str, new_string: str, file_path: str = N
target = skill_dir / "SKILL.md"
if not target.exists():
return _err(f"File not found: {target.relative_to(skill_dir)}")
target_label = file_path or "SKILL.md"
if read_guard := _background_review_read_before_write_guard(name, target, "patch", target_label):
return read_guard
content = target.read_text(encoding="utf-8")
# Same fuzzy engine as the file patch tool (whitespace/indent/escape
# normalization, block anchors) so minor formatting mismatches don't fail.
# Same fuzzy engine as the file patch tool (whitespace/indent/escape normalization,
# block anchors) so minor formatting mismatches don't fail.
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)
@@ -501,9 +486,8 @@ def _patch_skill(name: str, old_string: str, new_string: str, file_path: str = N
def _delete_skill(name: str, absorbed_into: Optional[str] = None) -> Dict[str, Any]:
"""Delete a skill. ``absorbed_into``: None = undeclared (legacy, accepted); "" =
explicit prune; "<skill>" = absorbed into that umbrella, which must exist on
disk (validated here so the model can't claim a nonexistent umbrella)."""
"""Delete a skill. ``absorbed_into``: None = undeclared (legacy, accepted); "" = explicit prune;
"<skill>" = absorbed into that umbrella, which must exist (so the model can't claim one)."""
skill_dir, guard = _locate_for_write(name, "delete")
if guard := guard or _curator_consolidation_delete_guard(name, absorbed_into):
return guard
@@ -521,9 +505,8 @@ def _delete_skill(name: str, absorbed_into: Optional[str] = None) -> Dict[str, A
skills_root = _containing_skills_root(skill_dir)
if unsafe := _validate_delete_target(skill_dir): # defense-in-depth before rmtree
return _err(unsafe)
# Curator consolidations must be RECOVERABLE (`hermes curator restore`): archive
# instead of rmtree. Foreground deletes keep hard-delete semantics.
# Curator consolidations must be RECOVERABLE (`hermes curator restore`): archive instead
# of rmtree. Foreground deletes keep hard-delete semantics.
absorbed_note = f" Content absorbed into '{absorbed_target}'." if absorbed_target else ""
if _is_background_review():
try:
@@ -533,9 +516,8 @@ def _delete_skill(name: str, absorbed_into: Optional[str] = None) -> Dict[str, A
return _err(f"failed to archive '{name}': {e}")
if not ok:
return _err(archive_msg)
return {"success": True,
"message": f"Skill '{name}' archived ({archive_msg}).{absorbed_note}",
"_archived": True}
return {"success": True, "_archived": True,
"message": f"Skill '{name}' archived ({archive_msg}).{absorbed_note}"}
shutil.rmtree(skill_dir)
_rmdir_if_empty(skill_dir.parent, skills_root) # empty category dir, never the root
@@ -582,7 +564,7 @@ def _remove_file(name: str, file_path: str) -> Dict[str, Any]:
target, err = _resolve_supporting_file(skill_dir, file_path)
if err:
return err
if not target.exists():
if not target.exists(): # list what IS there so the model can pick the right path
available = [
str(f.relative_to(skill_dir)) for subdir in ALLOWED_SUBDIRS
if (skill_dir / subdir).exists()
@@ -604,13 +586,11 @@ def _remove_file(name: str, file_path: str) -> Dict[str, Any]:
_skill_gate_bypass: "_ctxvars.ContextVar[bool]" = _ctxvars.ContextVar(
"skill_gate_bypass", default=False)
_GATED_ACTIONS = {"create", "edit", "patch", "delete", "write_file", "remove_file"}
def _run_write_gate(build_staging):
"""Shared write gate: None to proceed, else a JSON tool result (blocked/staged).
``build_staging(wa) -> (payload, gist)`` runs only when staging. Fails open
if write_approval cannot be imported."""
``build_staging(wa) -> (payload, gist)`` runs only when staging. Fails open if
write_approval cannot be imported."""
try:
from tools import write_approval as wa
except Exception:
@@ -627,9 +607,8 @@ def _run_write_gate(build_staging):
def _apply_skill_write_gate(action, name, **payload_kwargs):
"""Flat-shape gate: stage the full kwargs so approval can replay them;
bypassed during approved-pending replay."""
if action not in _GATED_ACTIONS or _skill_gate_bypass.get():
"""Flat-shape gate: stage the full kwargs so approval can replay them; bypassed during replay."""
if action not in _ACTION_HANDLERS or _skill_gate_bypass.get():
return None
def _staging(wa):
@@ -642,11 +621,12 @@ def _apply_skill_write_gate(action, name, **payload_kwargs):
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",
"absorbed_into", "operations")
def _skill_manage_from(payload: Dict[str, Any], **extra) -> str:
"""Call ``skill_manage`` with the flat-shape fields taken from ``payload``."""
"""Call ``skill_manage`` with the flat-shape fields (and absorbed_into/operations) of ``payload``."""
return skill_manage(
action=payload.get("action", ""), name=payload.get("name", ""),
replace_all=payload.get("replace_all", False),
@@ -657,9 +637,7 @@ def apply_skill_pending(payload: Dict[str, Any]) -> str:
"""Replay a staged skill write, bypassing the gate (the /skills approve handler)."""
token = _skill_gate_bypass.set(True)
try:
return _skill_manage_from(
payload, absorbed_into=payload.get("absorbed_into"),
operations=payload.get("operations"))
return _skill_manage_from(payload)
finally:
_skill_gate_bypass.reset(token)
@@ -672,9 +650,8 @@ _SYNC_PUSH_DEBOUNCE_S = 5.0
def _maybe_debounced_sync_push(skill_name: str) -> None:
"""Debounced best-effort sync push after a skill write; never blocks the caller.
Skills not opted into sync do nothing (no auth, no network); the push itself
(``skills_sync_client.maybe_push_skills``) enforces the access gate."""
"""Debounced best-effort sync push after a skill write; never blocks the caller. Skills not
opted into sync do nothing (no auth/network); ``maybe_push_skills`` enforces the access gate."""
global _sync_push_timer
try:
from tools.skill_usage import is_sync_enabled
@@ -733,20 +710,15 @@ _REQUIRED_ARGS = {
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."""
"""Best-effort post-mutation side effects (never break the tool): ledger, prompt-cache
clear, curator telemetry, debounced sync push."""
with suppress(Exception):
from tools import skill_ledger as _ledger
_post = _find_skill(name)
_evidence = {}
if action == "delete":
# consolidation vs prune, and whether the recoverable archive handled it
_evidence["absorbed_into"] = absorbed_into
_evidence["archived"] = bool(result.get("_archived"))
if session_id:
_evidence["session_id"] = session_id
if file_path:
_evidence["file_path"] = file_path
# delete: consolidation vs prune, and whether the recoverable archive handled it
_evidence = ({"absorbed_into": absorbed_into, "archived": bool(result.get("_archived"))}
if action == "delete" else {})
_evidence.update({k: v for k, v in (("session_id", session_id), ("file_path", file_path)) if v})
_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)
@@ -777,8 +749,8 @@ def skill_manage(
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:
"""Dispatch to the action handler; returns a JSON string. ``operations`` (batch
shape, applied atomically by _skill_manage_batch) overrides the flat fields."""
"""Dispatch to the action handler -> JSON string. ``operations`` (atomic batch shape,
see _skill_manage_batch) overrides the flat fields."""
if operations is not None:
return _skill_manage_batch(
operations, default_name=name or None, task_id=task_id, session_id=session_id)
@@ -794,9 +766,9 @@ def skill_manage(
if (gate_result := _apply_skill_write_gate(action, name, **args)) is not None:
return gate_result
# Audit ledger pre-capture: telemetry, not a gate — failures must NEVER block the
# 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 pre-capture: telemetry, not a gate — failures must NEVER block the mutation. delete
# destroys the whole package (consolidation may have re-homed support files first), so
# complete it from the newest curator backup or a restore is hollow.
_ledger_before = None
with suppress(Exception):
from tools import skill_ledger as _ledger
@@ -924,5 +896,4 @@ from tools.registry import registry, tool_error
registry.register(
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")))
args, task_id=kw.get("task_id"), session_id=kw.get("session_id")))