diff --git a/tools/skill_linter.py b/tools/skill_linter.py index 9692a175fe..31cfb59232 100644 --- a/tools/skill_linter.py +++ b/tools/skill_linter.py @@ -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") diff --git a/tools/skill_manager_batch.py b/tools/skill_manager_batch.py index 2e506692d1..92ad073119 100644 --- a/tools/skill_manager_batch.py +++ b/tools/skill_manager_batch.py @@ -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 diff --git a/tools/skill_manager_guards.py b/tools/skill_manager_guards.py index 713488e3f4..97712603b8 100644 --- a/tools/skill_manager_guards.py +++ b/tools/skill_manager_guards.py @@ -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= (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= (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 diff --git a/tools/skill_manager_tool.py b/tools/skill_manager_tool.py index 55da42deb5..b385560b31 100644 --- a/tools/skill_manager_tool.py +++ b/tools/skill_manager_tool.py @@ -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 `) 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"`) 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 . .' — 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")))