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

refactor(tools): skill_manage/ledger — shared identifier check, chainable result decorators, ledger read/filter fold

Browse Source
This commit is contained in:
Teknium
2026-09-03 01:30:26 -07:00
parent 370c6a2456
commit e6048ff6aa
4 changed files with 49 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
tools/skill_ledger.py
View File

@@ -63,18 +63,18 @@ def derive_actor() -> str:
return "agent"
def _skills_dir() -> Path:
return get_hermes_home() / "skills"
def ledger_path() -> Path:
return get_hermes_home() / "skills" / ".curator_ledger.jsonl"
return _skills_dir() / ".curator_ledger.jsonl"
def blobs_dir() -> Path:
return get_hermes_home() / ".curator_backups" / "blobs"
def _skills_dir() -> Path:
return get_hermes_home() / "skills"
def ledger_enabled() -> bool:
"""Config gate ``skills.ledger`` (default True); lazy import keeps this importable without the CLI."""
try:
@@ -85,14 +85,10 @@ def ledger_enabled() -> bool:
return True
def _norm(path: Path | str) -> Path:
return Path(os.path.normpath(str(path)))
def _rel_posix(path: Path | str, root: Path) -> Optional[str]:
"""POSIX path of ``path`` relative to ``root`` (both normalized), or None when outside."""
try:
return _norm(path).relative_to(_norm(root)).as_posix()
return Path(os.path.normpath(str(path))).relative_to(os.path.normpath(str(root))).as_posix()
except (ValueError, TypeError):
return None
@@ -118,11 +114,9 @@ def read_blob(sha256: str) -> Optional[bytes]:
"""Return blob content or None when missing/invalid."""
if not sha256 or not all(c in "0123456789abcdef" for c in sha256):
return None
try:
p = blobs_dir() / sha256
return p.read_bytes() if p.exists() else None
except OSError:
return None
with suppress(OSError):
return (blobs_dir() / sha256).read_bytes() if (blobs_dir() / sha256).exists() else None
return None
def snapshot_paths(root: Optional[Path], *, complete_package: bool = False) -> List[Dict[str, str]]:
@@ -132,15 +126,9 @@ def snapshot_paths(root: Optional[Path], *, complete_package: bool = False) -> L
tarball's files (disk hashes win)."""
if root is None:
return []
root = Path(root)
if root.is_file():
files = [root]
elif root.is_dir():
files = sorted(p for p in root.rglob("*") if p.is_file())
elif complete_package:
files = [] # gone from disk; the backup fill below may still recover it
else:
return []
root = Path(root) # gone from disk -> []; the complete_package fill may still recover it
files = ([root] if root.is_file()
else sorted(p for p in root.rglob("*") if p.is_file()) if root.is_dir() else [])
out = [{"path": str(f), "sha256": _store_blob(f.read_bytes())} for f in files]
return fill_snapshot_from_curator_backup(root, out) if complete_package else out
@@ -160,8 +148,7 @@ def _strip_archive_timestamp(name: str) -> str:
def _skill_md_parents(items: Optional[List[Dict[str, str]]]) -> List[Path]:
paths = [Path(str(item.get("path", ""))) for item in items or []]
return [p.parent for p in paths if p.name == "SKILL.md"]
return [p.parent for p in (Path(str(i.get("path", ""))) for i in items or []) if p.name == "SKILL.md"]
def package_prefixes(
@@ -310,9 +297,7 @@ def capture_before(
return None
try:
captured = snapshot_paths(root)
if complete_package:
captured = fill_snapshot_from_curator_backup(root, captured, skill=skill)
return captured
return fill_snapshot_from_curator_backup(root, captured, skill=skill) if complete_package else captured
except Exception as e:
logger.warning("skill_ledger: before-capture failed (%s) — mutation unaffected", e)
return None
@@ -328,18 +313,14 @@ def list_entries(skill: Optional[str] = None, limit: Optional[int] = None) -> Li
for line in lines:
with suppress(json.JSONDecodeError):
row = json.loads(line) if line.strip() else None
if isinstance(row, dict):
if isinstance(row, dict) and (not skill or row.get("skill") == skill):
rows.append(row)
if skill:
rows = [r for r in rows if r.get("skill") == skill]
rows.reverse()
return rows[:limit] if limit is not None and limit >= 0 else rows
def get_entry(entry_id: str) -> Optional[Dict[str, Any]]:
if not entry_id:
return None
return next((row for row in list_entries() if row.get("id") == entry_id), None)
return next((r for r in list_entries() if r.get("id") == entry_id), None) if entry_id else None
def _validate_entry_paths(entry: Dict[str, Any]) -> Optional[str]:

7
tools/skill_manager_batch.py
View File

@@ -34,8 +34,7 @@ def _validate_batch_ops(operations, default_name, tool_error):
names.append(nm)
if act == "create" and nm in names[:-1]:
return fail(i, f": create for '{nm}' must precede that skill's other ops.")
preflight = _background_review_preflight(act, nm)
if preflight is not None:
if (preflight := _background_review_preflight(act, nm)) is not None:
return None, json.dumps(preflight, ensure_ascii=False)
# Clobber guard: a DESTRUCTIVE op (create/write_file/remove_file/full rewrite) on
# a file an earlier op touched would SILENTLY discard its work — reject it.
@@ -156,8 +155,8 @@ def _skill_manage_batch(operations, default_name: str = None, task_id: str = Non
token = _smt._skill_gate_bypass.set(True)
try:
for i, op in enumerate(operations):
raw = _smt._skill_manage_from(
{**op, "name": names[i], "operations": None}, task_id=task_id, session_id=session_id)
raw = _smt._skill_manage_from({**op, "name": names[i], "operations": None},
task_id=task_id, session_id=session_id)
try:
parsed = json.loads(raw)
except Exception: # noqa: BLE001

8
tools/skill_manager_guards.py
View File

@@ -27,10 +27,9 @@ def _is_background_review() -> bool:
def _resolved_str(path: Path) -> str:
try:
with suppress(Exception):
return str(path.resolve())
except Exception:
return str(path)
return str(path)
class _BackgroundReviewReadMarks:
@@ -59,8 +58,7 @@ def mark_background_review_skill_read(path: Path) -> None:
the write guards require the mark."""
if not _is_background_review():
return
marks = _background_review_read_paths.get()
if marks is None:
if (marks := _background_review_read_paths.get()) is None:
_background_review_read_paths.set(marks := _BackgroundReviewReadMarks())
marks.add(_resolved_str(path))

65
tools/skill_manager_tool.py
View File

@@ -45,8 +45,7 @@ def _guard_agent_created_enabled() -> bool:
"""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(
cfg_get(load_config(), "skills", "guard_agent_created"), default=False)
return is_truthy_value(cfg_get(load_config(), "skills", "guard_agent_created"), default=False)
except Exception:
return False
@@ -102,14 +101,17 @@ def _display_create_dir() -> str:
# --- Validation helpers -------------------------------------------------------
def _check_identifier(value: str, label: str, invalid: str) -> Optional[str]:
if len(value) > MAX_NAME_LENGTH:
return f"{label} exceeds {MAX_NAME_LENGTH} characters."
return None if VALID_NAME_RE.match(value) else invalid
def _validate_name(name: str) -> Optional[str]:
if not name:
return "Skill name is required."
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}'. {_NAME_RULE} Must start with a letter or digit."
return None
return _check_identifier(
name, "Skill name", f"Invalid skill name '{name}'. {_NAME_RULE} Must start with a letter or digit.")
def _validate_category(category: Optional[str]) -> Optional[str]:
@@ -122,9 +124,7 @@ def _validate_category(category: Optional[str]) -> Optional[str]:
"Categories must be a single directory name.")
if "/" in category or "\\" in category:
return invalid
if len(category) > MAX_NAME_LENGTH:
return f"Category exceeds {MAX_NAME_LENGTH} characters."
return None if VALID_NAME_RE.match(category) else invalid
return _check_identifier(category, "Category", invalid)
def _validate_frontmatter(content: str, *, new_skill: bool = False) -> Optional[str]:
@@ -339,18 +339,20 @@ def _guarded_write(name: str, skill_dir: Path, target: Path, action: str, label:
return _err(scan_error)
def _attach_org_note(result: Dict[str, Any], name: str, skill_dir: Path) -> None:
def _attach_org_note(result: Dict[str, Any], name: str, skill_dir: Path) -> Dict[str, Any]:
if org_note := _maybe_auto_propose_org_edit(name, skill_dir):
result["org_sharing"] = org_note
result["message"] = f"{result['message']} {org_note}"
return result
def _add_description_prompt_preview(result: Dict[str, Any], content: str) -> None:
def _add_description_prompt_preview(result: Dict[str, Any], content: str) -> Dict[str, Any]:
fm, _ = _parse_frontmatter(content)
if is_skill_description_truncated_for_prompt(fm):
result["system_prompt_preview"] = (
f"System prompt will show: \"{extract_skill_description(fm)}\" — keep the trigger "
f"self-contained in the first {SKILL_PROMPT_DESC_LIMIT - 3} chars.")
return result
def _attach_lint_findings(result: Dict[str, Any], skill_md: Path) -> None:
@@ -376,9 +378,8 @@ def _clip(text: str, n: int, ellipsis: str) -> str:
# --- Core actions -------------------------------------------------------------
def _create_skill(name: str, content: str, category: str = None) -> Dict[str, Any]:
err = (_validate_name(name) or _validate_category(category)
or _validate_frontmatter(content, new_skill=True) or _validate_content_size(content))
if err:
if err := (_validate_name(name) or _validate_category(category)
or _validate_frontmatter(content, new_skill=True) or _validate_content_size(content)):
return _err(err)
if existing := _find_skill(name):
return _err(f"A skill named '{name}' already exists at {existing['path']}.")
@@ -389,8 +390,7 @@ def _create_skill(name: str, content: str, category: str = None) -> Dict[str, An
if scan_error := _security_scan_skill(skill_dir):
shutil.rmtree(skill_dir, ignore_errors=True)
return _err(scan_error)
root = _skills_dir()
# Relative when under the profile dir; absolute when created under skills.create_dir.
root = _skills_dir() # display relative under the profile dir; absolute under skills.create_dir
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),
@@ -399,8 +399,7 @@ def _create_skill(name: str, content: str, category: str = None) -> Dict[str, An
"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)
_attach_lint_findings(_add_description_prompt_preview(result, content), skill_md)
return result
@@ -410,15 +409,12 @@ def _edit_skill(name: str, content: str) -> Dict[str, Any]:
return _err(err)
skill_dir, guard = _locate_for_write(name, "edit")
# SKILL.md always exists here (_find_skill requires it), so a blocked scan restores it.
if guard := guard or _guarded_write(
name, skill_dir, skill_dir / "SKILL.md", "edit", "SKILL.md", content):
if guard := guard or _guarded_write(name, skill_dir, skill_dir / "SKILL.md", "edit", "SKILL.md", content):
return guard
result = {
"success": True, "message": f"Skill '{name}' updated (full rewrite).",
"path": str(skill_dir), "_change": {"description": _description_preview(content)}}
_attach_org_note(result, name, skill_dir)
_add_description_prompt_preview(result, content)
return result
return _add_description_prompt_preview(_attach_org_note(result, name, skill_dir), content)
def _patch_skill(name: str, old_string: str, new_string: str, file_path: str = None,
@@ -471,8 +467,7 @@ def _patch_skill(name: str, old_string: str, new_string: str, file_path: str = N
"success": True,
"message": f"Patched {target_label} in skill '{name}' ({match_count} replacement{'s' if match_count > 1 else ''}).",
"_change": {"old": _clip(old_string, 200, "…"), "new": _clip(new_string, 200, "…")}}
_attach_org_note(result, name, skill_dir)
return result
return _attach_org_note(result, name, skill_dir)
def _delete_skill(name: str, absorbed_into: Optional[str] = None) -> Dict[str, Any]:
@@ -523,8 +518,7 @@ def _write_file(name: str, file_path: str, file_content: str) -> Dict[str, Any]:
return _err(err)
if not file_content and file_content != "":
return _err("file_content is required.")
content_bytes = len(file_content.encode("utf-8"))
if content_bytes > MAX_SKILL_FILE_BYTES:
if (content_bytes := len(file_content.encode("utf-8"))) > MAX_SKILL_FILE_BYTES:
return _err(f"File content is {content_bytes:,} bytes (limit: {MAX_SKILL_FILE_BYTES:,} "
f"bytes / 1 MiB). Consider splitting into smaller files.")
if err := _validate_content_size(file_content, label=file_path):
@@ -535,10 +529,8 @@ def _write_file(name: str, file_path: str, file_content: str) -> Dict[str, Any]:
target, err = _resolve_supporting_file(skill_dir, file_path)
if guard := err or _guarded_write(name, skill_dir, target, "write_file", file_path, file_content):
return guard
result = {"success": True, "message": f"File '{file_path}' written to skill '{name}'.",
"path": str(target)}
_attach_org_note(result, name, skill_dir)
return result
return _attach_org_note({"success": True, "message": f"File '{file_path}' written to skill '{name}'.",
"path": str(target)}, name, skill_dir)
def _remove_file(name: str, file_path: str) -> Dict[str, Any]:
@@ -552,12 +544,9 @@ def _remove_file(name: str, file_path: str) -> Dict[str, Any]:
if err:
return err
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()
for f in (skill_dir / subdir).rglob("*") 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 = [str(f.relative_to(skill_dir)) for subdir in ALLOWED_SUBDIRS
if (skill_dir / subdir).exists() for f in (skill_dir / subdir).rglob("*") if f.is_file()]
return _err(f"File '{file_path}' not found in skill '{name}'.", available_files=available or None)
if read_guard := _background_review_read_before_write_guard(name, target, "remove_file", file_path):
return read_guard
target.unlink()