diff --git a/tools/skill_ledger.py b/tools/skill_ledger.py index ffe617b7ee..8b4512ee4f 100644 --- a/tools/skill_ledger.py +++ b/tools/skill_ledger.py @@ -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//``, 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//``, 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}'" diff --git a/tools/skill_manager_batch.py b/tools/skill_manager_batch.py index c0a2f4eb8c..79fa30341c 100644 --- a/tools/skill_manager_batch.py +++ b/tools/skill_manager_batch.py @@ -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 diff --git a/tools/skill_manager_guards.py b/tools/skill_manager_guards.py index 41e8a02b39..143c308238 100644 --- a/tools/skill_manager_guards.py +++ b/tools/skill_manager_guards.py @@ -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=`` (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=`` (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: diff --git a/tools/skill_manager_tool.py b/tools/skill_manager_tool.py index e1c45f3f49..d99d977435 100644 --- a/tools/skill_manager_tool.py +++ b/tools/skill_manager_tool.py @@ -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: ``/[category/]/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: +``/[category/]/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; "" = 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; + "" = 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")))