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

refactor(tools): skill guards/tool — is_relative_to collapses, string reflow

Browse Source
This commit is contained in:
Teknium
2026-09-02 23:03:24 -07:00
parent b187d4e547
commit a861976519
4 changed files with 91 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/skill_linter.py
+1 -2
View File
@@ -100,8 +100,7 @@ def _check_frontmatter(frontmatter: Dict[str, Any], skill_dir: Optional[Path]) -
yield _warn("missing-metadata", "metadata.hermes.tags is missing.")
author = str(frontmatter.get("author", ""))
if author and author.strip().lower() in ("hermes", "agent", "hermes agent") and (
author != "Hermes Agent"
):
author != "Hermes Agent"):
yield _warn("author-caps", f"author '{author}' should be 'Hermes Agent' (proper caps) "
f"or a real contributor name.")
platforms = frontmatter.get("platforms")
tools/skill_manager_batch.py
+6 -8
View File
@@ -31,9 +31,8 @@ def _validate_batch_ops(operations, default_name, tool_error):
act = op["action"]
if act not in _BATCH_OP_ACTIONS:
return None, tool_error(
f"operations[{i}]: unknown action '{act}'. "
f"Batchable: {', '.join(sorted(_BATCH_OP_ACTIONS))}; "
"delete must be sole.",
f"operations[{i}]: unknown action '{act}'. Batchable: "
f"{', '.join(sorted(_BATCH_OP_ACTIONS))}; delete must be sole.",
success=False)
nm = op.get("name") or default_name
if not nm:
@@ -61,11 +60,10 @@ def _validate_batch_ops(operations, default_name, tool_error):
destructive = act in ("create", "write_file", "remove_file") or full_rewrite
if destructive and key in touched_files:
return None, tool_error(
f"operations[{i}]: {act} on '{target}' of skill '{nm}' — an "
"earlier op in this batch already touched that file, and this "
"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.",
f"operations[{i}]: {act} on '{target}' of skill '{nm}' — an earlier op in this "
f"batch already touched that file, and this op would silently discard its work. "
f"One destructive op (write_file/remove_file/full rewrite) per file per batch; put "
f"it first, or fold the change in. Patch chains are fine.",
success=False)
touched_files.add(key)
return names, None
tools/skill_manager_guards.py
+34 -50
View File
@@ -65,8 +65,7 @@ def mark_background_review_skill_read(path: Path) -> None:
return
marks = _background_review_read_paths.get()
if marks is None:
marks = _BackgroundReviewReadMarks()
_background_review_read_paths.set(marks)
_background_review_read_paths.set(marks := _BackgroundReviewReadMarks())
marks.add(_resolved_str(path))
@@ -93,9 +92,9 @@ def _containing_skills_root(skill_path: Path) -> Path:
resolved = skill_path
for root in get_all_skills_dirs():
try:
resolved.relative_to(root.resolve())
return root
except (ValueError, OSError):
if resolved.is_relative_to(root.resolve()):
return root
except OSError:
continue
return _smt._skills_dir()
@@ -133,15 +132,11 @@ def _validate_delete_target(skill_dir: Path) -> Optional[str]:
return (
f"Refusing to delete '{skill_dir}': resolves to the skills root "
f"itself, which would remove every installed skill.")
try:
rel = resolved.relative_to(root)
except ValueError:
continue
if rel.parts:
if resolved.is_relative_to(root):
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 --------------------------------------------
@@ -165,19 +160,16 @@ def _pinned_guard(name: str) -> Optional[str]:
from tools import skill_usage
if skill_usage.get_record(name).get("pinned"):
return (
f"Skill '{name}' is pinned and cannot be deleted by "
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"Skill '{name}' is pinned and cannot be deleted by skill_manage. Ask the user to "
f"run `hermes curator unpin {name}` if they want to delete it. Patches and edits "
f"are allowed on pinned skills; only 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,
) -> Optional[Dict[str, Any]]:
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
@@ -189,10 +181,9 @@ def _background_review_write_guard(
from tools import skill_usage
if skill_usage.get_record(name).get("pinned"):
return _refusal(
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"Refusing background curator {action} for pinned skill '{name}': pinned skills "
f"are off-limits to autonomous maintenance. Ask the user to run `hermes curator "
f"unpin {name}` if they want it changed.")
except Exception:
logger.debug("pinned skill guard lookup failed for %s", name, exc_info=True)
@@ -214,7 +205,7 @@ def _background_review_write_guard(
(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} skill '{name}'.")
# Not curator-managed (no `created_by: "agent"`) => user-owned. A MISSING
# record and an explicit `created_by: null` must resolve IDENTICALLY (keying
# on presence made the policy depend on the guard's own side effect: the
@@ -224,16 +215,14 @@ def _background_review_write_guard(
_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"Refusing background curator {action} for skill '{name}': the skill is not "
f"curator-managed ({_detail}). User-owned skills are off-limits to autonomous "
f"curation. 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.")
f"Refusing background curator {action} for skill '{name}': agent ownership could not "
f"be verified because the provenance record is unavailable or unreadable.")
return None
@@ -243,11 +232,10 @@ def _background_review_read_before_write_guard(
if not _is_background_review() or _background_review_has_read(target):
return None
return _refusal(
f"Refusing background curator {action} for skill '{name}': "
f"the current {file_label} content has not been loaded in this "
"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.",
f"Refusing background curator {action} for skill '{name}': the current {file_label} "
f"content has not been loaded in this review turn. Call skill_view(name) for SKILL.md, or "
f"skill_view(name, file_path=...) for a supporting file, then retry the write using the "
f"content just returned.",
_read_before_write_required=True)
@@ -260,8 +248,7 @@ 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]]:
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
@@ -273,12 +260,11 @@ def _curator_consolidation_delete_guard(
if isinstance(absorbed_into, str) and absorbed_into.strip():
return None
return _refusal(
f"Refusing background curator delete of skill '{name}': the "
"consolidation pass may only archive a skill it has absorbed into "
"an umbrella. Pass absorbed_into=<umbrella> (the umbrella must "
"already exist) to record a verified consolidation. Pruning a "
"skill with no forwarding target is not permitted here — the "
"deterministic inactivity prune handles staleness archival "
f"Refusing background curator delete of skill '{name}': the consolidation pass may only "
f"archive a skill it has absorbed into an umbrella. Pass absorbed_into=<umbrella> (the "
f"umbrella must already exist) to record a verified consolidation. Pruning a skill with no "
f"forwarding target is not permitted here — the deterministic inactivity prune handles "
f"staleness archival "
"separately. Keeping '{name}' active.".format(name=name),
_fail_closed=True)
@@ -331,12 +317,10 @@ def _org_mirror_write_guard(name: str, skill_path: Path, action: str) -> Optiona
if is_org_mirror_path(skill_path, _smt._skills_dir()):
return _refusal(
f"Cannot {action} '{name}' locally: it is shared by your "
"organisation, so a local delete would just come back on "
"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"Cannot {action} '{name}' locally: it is shared by your organisation, so a local "
f"delete would just come back on the next sync. Ask an org admin to remove it for "
f"everyone. (Editing it IS allowed — your changes are kept and can be proposed "
f"back with `hermes sync propose {name}`.)")
except Exception:
logger.debug("org mirror guard lookup failed for %s", name, exc_info=True)
return None
tools/skill_manager_tool.py
+50 -44
View File
@@ -34,8 +34,7 @@ from tools.skill_manager_guards import ( # noqa: F401 — re-exported for calle
_background_review_write_guard, _containing_skills_root, _curator_consolidation_delete_guard,
_is_path_redirect, _maybe_auto_propose_org_edit, _org_mirror_write_guard, _pinned_guard,
_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)
from tools.skills_guard import scan_skill, should_allow_install, format_scan_report
@@ -243,11 +242,9 @@ def _find_skill(name: str) -> Optional[Dict[str, Any]]:
if skill_dir.name == name:
return {"path": skill_dir}
if local_root is not None:
try:
rel = skill_dir.resolve().relative_to(local_root)
except ValueError:
continue
if rel.as_posix() == name: # POSIX form so it works on Windows too
resolved = skill_dir.resolve()
if (resolved.is_relative_to(local_root)
and resolved.relative_to(local_root).as_posix() == name): # POSIX form
return {"path": skill_dir}
return None
@@ -301,9 +298,8 @@ def _skill_not_found_error(name: str, suffix: str = "") -> str:
elif others:
names = ", ".join(f"'{p}'" for p, _ in others)
base += (
f" Skills by that name exist in other profiles: {names}. Switch profiles "
f"(`hermes -p <name>`) to edit there, or edit the files directly "
f"(file tools / terminal).")
f" Skills by that name exist in other profiles: {names}. Switch profiles (`hermes -p "
f"<name>`) to edit there, or edit the files directly (file tools / terminal).")
else:
base += " Use skills_list() to see available skills."
return base + suffix
@@ -433,12 +429,11 @@ def _create_skill(name: str, content: str, category: str = None) -> Dict[str, An
shutil.rmtree(skill_dir, ignore_errors=True)
return _err(scan_error)
try:
_display_path = str(skill_dir.relative_to(_skills_dir()))
except ValueError:
_display_path = str(skill_dir) # created under skills.create_dir
root = _skills_dir()
# Relative when under the profile dir; absolute when created 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": _display_path,
"success": True, "message": f"Skill '{name}' created.", "path": str(display),
"skill_md": str(skill_md), "_change": {"description": _description_preview(content)},
}
if category:
@@ -466,8 +461,7 @@ def _edit_skill(name: str, content: str) -> Dict[str, Any]:
return guard
result = {
"success": True, "message": f"Skill '{name}' updated (full rewrite).",
"path": str(skill_dir), "_change": {"description": _description_preview(content)},
}
"path": str(skill_dir), "_change": {"description": _description_preview(content)}}
_attach_org_note(result, name, skill_dir)
_add_description_prompt_preview(result, content)
return result
@@ -481,11 +475,10 @@ def _patch_skill(name: str, old_string: str, new_string: str, file_path: str = N
# 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 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.")
"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 "
"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.")
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
@@ -535,8 +528,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": 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
@@ -556,8 +548,7 @@ def _delete_skill(name: str, absorbed_into: Optional[str] = None) -> Dict[str, A
return _err(pinned_err)
absorbed_target = absorbed_into.strip() if isinstance(absorbed_into, str) else ""
is_consolidation = bool(absorbed_target)
if is_consolidation:
if absorbed_target:
if absorbed_target == name:
return _err(f"absorbed_into='{absorbed_target}' cannot equal the skill being deleted.")
if not _find_skill(absorbed_target):
@@ -571,7 +562,7 @@ def _delete_skill(name: str, absorbed_into: Optional[str] = None) -> Dict[str, A
# 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 is_consolidation else ""
absorbed_note = f" Content absorbed into '{absorbed_target}'." if absorbed_target else ""
if _is_background_review():
try:
from tools.skill_usage import archive_skill
@@ -783,8 +774,7 @@ _ACTION_HANDLERS = {
"patch": _act_patch,
"delete": lambda a: _delete_skill(a["name"], absorbed_into=a["absorbed_into"]),
"write_file": lambda a: _write_file(a["name"], a["file_path"], a["file_content"]),
"remove_file": lambda a: _remove_file(a["name"], a["file_path"]),
}
"remove_file": lambda a: _remove_file(a["name"], a["file_path"])}
# action -> (arg, is_missing, error) argument-shape checks run before the handler.
_REQUIRED_ARGS = {
"create": [("content", lambda v: not v,
@@ -794,8 +784,7 @@ _REQUIRED_ARGS = {
"write_file": [("file_path", lambda v: not v,
"file_path is required for 'write_file'. Example: 'references/api-guide.md'"),
("file_content", lambda v: v is None, "file_content is required for 'write_file'.")],
"remove_file": [("file_path", lambda v: not v, "file_path is required for 'remove_file'.")],
}
"remove_file": [("file_path", lambda v: not v, "file_path is required for 'remove_file'.")]}
def _record_success(action, name, result, *, file_path, absorbed_into, task_id,
@@ -910,7 +899,8 @@ 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": {
@@ -925,19 +915,25 @@ 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": {
@@ -946,10 +942,12 @@ 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": (
@@ -958,15 +956,24 @@ 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"]
}
},
# Also accepted, never advertised: the legacy flat single-op fields, and
# `absorbed_into` on delete ops (curator-only vocabulary; the curator's
# prompt documents it and the delete guard's error re-teaches it).
},
"required": ["operations"]}}
"required": ["operations"],
},
}
# --- Registry ---
@@ -976,5 +983,4 @@ 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")),
)
task_id=kw.get("task_id"), session_id=kw.get("session_id")))