fix: Update tools/skill_manager_tool.py with rich error context (#624)
This commit is contained in:
@@ -40,7 +40,7 @@ import shutil
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from hermes_constants import get_hermes_home
|
||||
from typing import Dict, Any, Optional
|
||||
from typing import Dict, Any, Optional, Tuple
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -53,6 +53,57 @@ except ImportError:
|
||||
_GUARD_AVAILABLE = False
|
||||
|
||||
|
||||
|
||||
def _format_error(error_msg: str, skill_name: str = None, file_path: str = None,
|
||||
suggestion: str = None, context: dict = None) -> dict:
|
||||
"""Format an error response with rich context for debugging.
|
||||
|
||||
Args:
|
||||
error_msg: The primary error message
|
||||
skill_name: Name of the skill being operated on
|
||||
file_path: Path to the file that failed
|
||||
suggestion: Suggested action to fix the issue
|
||||
context: Additional context dict (e.g., {'line': 5, 'found': 'x', 'expected': 'y'})
|
||||
|
||||
Returns:
|
||||
Formatted error dict with success=False
|
||||
"""
|
||||
parts = [error_msg]
|
||||
|
||||
if skill_name:
|
||||
parts.append(f"Skill: {skill_name}")
|
||||
|
||||
if file_path:
|
||||
parts.append(f"File: {file_path}")
|
||||
|
||||
if context:
|
||||
for key, value in context.items():
|
||||
parts.append(f"{key}: {value}")
|
||||
|
||||
if suggestion:
|
||||
parts.append(f"Suggestion: {suggestion}")
|
||||
|
||||
return {
|
||||
"success": False,
|
||||
"error": " | ".join(parts),
|
||||
"skill_name": skill_name,
|
||||
"file_path": file_path,
|
||||
"suggestion": suggestion,
|
||||
}
|
||||
|
||||
|
||||
def _get_skill_file_path(skill_name: str, file_path: str = None) -> str:
|
||||
"""Get the full file path for error messages."""
|
||||
existing = _find_skill(skill_name)
|
||||
if not existing:
|
||||
return f"(skill '{skill_name}' not found)"
|
||||
|
||||
skill_dir = existing["path"]
|
||||
if file_path:
|
||||
return str(skill_dir / file_path)
|
||||
return str(skill_dir / "SKILL.md")
|
||||
|
||||
|
||||
def _security_scan_skill(skill_dir: Path) -> Optional[str]:
|
||||
"""Scan a skill directory after write. Returns error string if blocked, else None."""
|
||||
if not _GUARD_AVAILABLE:
|
||||
@@ -92,11 +143,6 @@ VALID_NAME_RE = re.compile(r'^[a-z0-9][a-z0-9._-]*$')
|
||||
ALLOWED_SUBDIRS = {"references", "templates", "scripts", "assets"}
|
||||
|
||||
|
||||
def check_skill_manage_requirements() -> bool:
|
||||
"""Skill management has no external requirements -- always available."""
|
||||
return True
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Validation helpers
|
||||
# =============================================================================
|
||||
@@ -146,10 +192,10 @@ def _validate_frontmatter(content: str) -> Optional[str]:
|
||||
Returns error message or None if valid.
|
||||
"""
|
||||
if not content.strip():
|
||||
return "Content cannot be empty."
|
||||
return "Content cannot be empty. SKILL.md must have YAML frontmatter and body content."
|
||||
|
||||
if not content.startswith("---"):
|
||||
return "SKILL.md must start with YAML frontmatter (---). See existing skills for format."
|
||||
return "SKILL.md must start with YAML frontmatter delimiter (---). Expected format: ---\nname: skill-name\ndescription: ...\n---\n<body content>"
|
||||
|
||||
end_match = re.search(r'\n---\s*\n', content[3:])
|
||||
if not end_match:
|
||||
@@ -163,7 +209,7 @@ def _validate_frontmatter(content: str) -> Optional[str]:
|
||||
return f"YAML frontmatter parse error: {e}"
|
||||
|
||||
if not isinstance(parsed, dict):
|
||||
return "Frontmatter must be a YAML mapping (key: value pairs)."
|
||||
return "Frontmatter must be a YAML mapping (key: value pairs). Check for syntax errors in the YAML."
|
||||
|
||||
if "name" not in parsed:
|
||||
return "Frontmatter must include 'name' field."
|
||||
@@ -224,13 +270,15 @@ def _validate_file_path(file_path: str) -> Optional[str]:
|
||||
Validate a file path for write_file/remove_file.
|
||||
Must be under an allowed subdirectory and not escape the skill dir.
|
||||
"""
|
||||
from tools.path_security import has_traversal_component
|
||||
|
||||
if not file_path:
|
||||
return "file_path is required."
|
||||
|
||||
normalized = Path(file_path)
|
||||
|
||||
# Prevent path traversal
|
||||
if ".." in normalized.parts:
|
||||
if has_traversal_component(file_path):
|
||||
return "Path traversal ('..') is not allowed."
|
||||
|
||||
# Must be under an allowed subdirectory
|
||||
@@ -245,6 +293,17 @@ def _validate_file_path(file_path: str) -> Optional[str]:
|
||||
return None
|
||||
|
||||
|
||||
def _resolve_skill_target(skill_dir: Path, file_path: str) -> Tuple[Optional[Path], Optional[str]]:
|
||||
"""Resolve a supporting-file path and ensure it stays within the skill directory."""
|
||||
from tools.path_security import validate_within_dir
|
||||
|
||||
target = skill_dir / file_path
|
||||
error = validate_within_dir(target, skill_dir)
|
||||
if error:
|
||||
return None, error
|
||||
return target, None
|
||||
|
||||
|
||||
def _atomic_write_text(file_path: Path, content: str, encoding: str = "utf-8") -> None:
|
||||
"""
|
||||
Atomically write text content to a file.
|
||||
@@ -304,10 +363,12 @@ def _create_skill(name: str, content: str, category: str = None) -> Dict[str, An
|
||||
# Check for name collisions across all directories
|
||||
existing = _find_skill(name)
|
||||
if existing:
|
||||
return {
|
||||
"success": False,
|
||||
"error": f"A skill named '{name}' already exists at {existing['path']}."
|
||||
}
|
||||
return _format_error(
|
||||
f"A skill named '{name}' already exists",
|
||||
skill_name=name,
|
||||
file_path=str(existing['path']),
|
||||
suggestion="Use skill_manage(action='edit') to update the existing skill or choose a different name"
|
||||
)
|
||||
|
||||
# Create the skill directory
|
||||
skill_dir = _resolve_skill_dir(name, category)
|
||||
@@ -321,7 +382,12 @@ def _create_skill(name: str, content: str, category: str = None) -> Dict[str, An
|
||||
scan_error = _security_scan_skill(skill_dir)
|
||||
if scan_error:
|
||||
shutil.rmtree(skill_dir, ignore_errors=True)
|
||||
return {"success": False, "error": scan_error}
|
||||
return _format_error(
|
||||
scan_error,
|
||||
skill_name=name,
|
||||
file_path=str(skill_dir),
|
||||
suggestion="Review the security scan report and fix flagged issues"
|
||||
)
|
||||
|
||||
result = {
|
||||
"success": True,
|
||||
@@ -350,7 +416,11 @@ def _edit_skill(name: str, content: str) -> Dict[str, Any]:
|
||||
|
||||
existing = _find_skill(name)
|
||||
if not existing:
|
||||
return {"success": False, "error": f"Skill '{name}' not found. Use skills_list() to see available skills."}
|
||||
return _format_error(
|
||||
f"Skill '{name}' not found",
|
||||
skill_name=name,
|
||||
suggestion="Use skills_list() to see available skills or skill_manage(action='create') to create it"
|
||||
)
|
||||
|
||||
skill_md = existing["path"] / "SKILL.md"
|
||||
# Back up original content for rollback
|
||||
@@ -362,7 +432,12 @@ def _edit_skill(name: str, content: str) -> Dict[str, Any]:
|
||||
if scan_error:
|
||||
if original_content is not None:
|
||||
_atomic_write_text(skill_md, original_content)
|
||||
return {"success": False, "error": scan_error}
|
||||
return _format_error(
|
||||
scan_error,
|
||||
skill_name=name,
|
||||
file_path=str(skill_md),
|
||||
suggestion="Review the security scan report and fix flagged issues"
|
||||
)
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
@@ -384,13 +459,23 @@ def _patch_skill(
|
||||
Requires a unique match unless replace_all is True.
|
||||
"""
|
||||
if not old_string:
|
||||
return {"success": False, "error": "old_string is required for 'patch'."}
|
||||
return _format_error(
|
||||
"old_string is required for 'patch'",
|
||||
suggestion="Provide the text to find in the skill file. Use skill_manage(action='edit') for full rewrites."
|
||||
)
|
||||
if new_string is None:
|
||||
return {"success": False, "error": "new_string is required for 'patch'. Use an empty string to delete matched text."}
|
||||
return _format_error(
|
||||
"new_string is required for 'patch'",
|
||||
suggestion="Provide the replacement text. Use empty string '' to delete the matched text."
|
||||
)
|
||||
|
||||
existing = _find_skill(name)
|
||||
if not existing:
|
||||
return {"success": False, "error": f"Skill '{name}' not found."}
|
||||
return _format_error(
|
||||
f"Skill '{name}' not found",
|
||||
skill_name=name,
|
||||
suggestion="Use skills_list() to see available skills"
|
||||
)
|
||||
|
||||
skill_dir = existing["path"]
|
||||
|
||||
@@ -399,13 +484,20 @@ def _patch_skill(
|
||||
err = _validate_file_path(file_path)
|
||||
if err:
|
||||
return {"success": False, "error": err}
|
||||
target = skill_dir / file_path
|
||||
target, err = _resolve_skill_target(skill_dir, file_path)
|
||||
if err:
|
||||
return {"success": False, "error": err}
|
||||
else:
|
||||
# Patching SKILL.md
|
||||
target = skill_dir / "SKILL.md"
|
||||
|
||||
if not target.exists():
|
||||
return {"success": False, "error": f"File not found: {target.relative_to(skill_dir)}"}
|
||||
return _format_error(
|
||||
f"File not found: {target.relative_to(skill_dir)}",
|
||||
skill_name=name,
|
||||
file_path=str(target),
|
||||
suggestion=f"Check the file path. Available files in skill: {list(skill_dir.glob('**/*'))}"
|
||||
)
|
||||
|
||||
content = target.read_text(encoding="utf-8")
|
||||
|
||||
@@ -415,17 +507,19 @@ def _patch_skill(
|
||||
# from exact-match failures on minor formatting mismatches.
|
||||
from tools.fuzzy_match import fuzzy_find_and_replace
|
||||
|
||||
new_content, match_count, match_error = fuzzy_find_and_replace(
|
||||
new_content, match_count, _strategy, match_error = fuzzy_find_and_replace(
|
||||
content, old_string, new_string, replace_all
|
||||
)
|
||||
if match_error:
|
||||
# Show a short preview of the file so the model can self-correct
|
||||
preview = content[:500] + ("..." if len(content) > 500 else "")
|
||||
return {
|
||||
"success": False,
|
||||
"error": match_error,
|
||||
"file_preview": preview,
|
||||
}
|
||||
return _format_error(
|
||||
f"Pattern match failed: {match_error}",
|
||||
skill_name=name,
|
||||
file_path=str(target),
|
||||
context={"file_preview": preview[:200] + "..." if len(preview) > 200 else preview},
|
||||
suggestion="Check for whitespace differences, indentation, or escaping issues in old_string"
|
||||
)
|
||||
|
||||
# Check size limit on the result
|
||||
target_label = "SKILL.md" if not file_path else file_path
|
||||
@@ -437,10 +531,12 @@ def _patch_skill(
|
||||
if not file_path:
|
||||
err = _validate_frontmatter(new_content)
|
||||
if err:
|
||||
return {
|
||||
"success": False,
|
||||
"error": f"Patch would break SKILL.md structure: {err}",
|
||||
}
|
||||
return _format_error(
|
||||
f"Patch would break SKILL.md structure: {err}",
|
||||
skill_name=name,
|
||||
file_path=str(target),
|
||||
suggestion="Ensure the patch doesn't corrupt YAML frontmatter (--- delimiters and key: value format)"
|
||||
)
|
||||
|
||||
original_content = content # for rollback
|
||||
_atomic_write_text(target, new_content)
|
||||
@@ -461,7 +557,11 @@ def _delete_skill(name: str) -> Dict[str, Any]:
|
||||
"""Delete a skill."""
|
||||
existing = _find_skill(name)
|
||||
if not existing:
|
||||
return {"success": False, "error": f"Skill '{name}' not found."}
|
||||
return _format_error(
|
||||
f"Skill '{name}' not found",
|
||||
skill_name=name,
|
||||
suggestion="Use skills_list() to see available skills"
|
||||
)
|
||||
|
||||
skill_dir = existing["path"]
|
||||
shutil.rmtree(skill_dir)
|
||||
@@ -503,9 +603,15 @@ def _write_file(name: str, file_path: str, file_content: str) -> Dict[str, Any]:
|
||||
|
||||
existing = _find_skill(name)
|
||||
if not existing:
|
||||
return {"success": False, "error": f"Skill '{name}' not found. Create it first with action='create'."}
|
||||
return _format_error(
|
||||
f"Skill '{name}' not found",
|
||||
skill_name=name,
|
||||
suggestion="Use skills_list() to see available skills"
|
||||
)
|
||||
|
||||
target = existing["path"] / file_path
|
||||
target, err = _resolve_skill_target(existing["path"], file_path)
|
||||
if err:
|
||||
return {"success": False, "error": err}
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
# Back up for rollback
|
||||
original_content = target.read_text(encoding="utf-8") if target.exists() else None
|
||||
@@ -535,10 +641,16 @@ def _remove_file(name: str, file_path: str) -> Dict[str, Any]:
|
||||
|
||||
existing = _find_skill(name)
|
||||
if not existing:
|
||||
return {"success": False, "error": f"Skill '{name}' not found."}
|
||||
return _format_error(
|
||||
f"Skill '{name}' not found",
|
||||
skill_name=name,
|
||||
suggestion="Use skills_list() to see available skills"
|
||||
)
|
||||
skill_dir = existing["path"]
|
||||
|
||||
target = skill_dir / file_path
|
||||
target, err = _resolve_skill_target(skill_dir, file_path)
|
||||
if err:
|
||||
return {"success": False, "error": err}
|
||||
if not target.exists():
|
||||
# List what's actually there for the model to see
|
||||
available = []
|
||||
@@ -589,19 +701,19 @@ def skill_manage(
|
||||
"""
|
||||
if action == "create":
|
||||
if not content:
|
||||
return json.dumps({"success": False, "error": "content is required for 'create'. Provide the full SKILL.md text (frontmatter + body)."}, ensure_ascii=False)
|
||||
return tool_error("content is required for 'create'. Provide the full SKILL.md text (frontmatter + body).", success=False)
|
||||
result = _create_skill(name, content, category)
|
||||
|
||||
elif action == "edit":
|
||||
if not content:
|
||||
return json.dumps({"success": False, "error": "content is required for 'edit'. Provide the full updated SKILL.md text."}, ensure_ascii=False)
|
||||
return tool_error("content is required for 'edit'. Provide the full updated SKILL.md text.", success=False)
|
||||
result = _edit_skill(name, content)
|
||||
|
||||
elif action == "patch":
|
||||
if not old_string:
|
||||
return json.dumps({"success": False, "error": "old_string is required for 'patch'. Provide the text to find."}, ensure_ascii=False)
|
||||
return tool_error("old_string is required for 'patch'. Provide the text to find.", success=False)
|
||||
if new_string is None:
|
||||
return json.dumps({"success": False, "error": "new_string is required for 'patch'. Use empty string to delete matched text."}, ensure_ascii=False)
|
||||
return tool_error("new_string is required for 'patch'. Use empty string to delete matched text.", success=False)
|
||||
result = _patch_skill(name, old_string, new_string, file_path, replace_all)
|
||||
|
||||
elif action == "delete":
|
||||
@@ -609,14 +721,14 @@ def skill_manage(
|
||||
|
||||
elif action == "write_file":
|
||||
if not file_path:
|
||||
return json.dumps({"success": False, "error": "file_path is required for 'write_file'. Example: 'references/api-guide.md'"}, ensure_ascii=False)
|
||||
return tool_error("file_path is required for 'write_file'. Example: 'references/api-guide.md'", success=False)
|
||||
if file_content is None:
|
||||
return json.dumps({"success": False, "error": "file_content is required for 'write_file'."}, ensure_ascii=False)
|
||||
return tool_error("file_content is required for 'write_file'.", success=False)
|
||||
result = _write_file(name, file_path, file_content)
|
||||
|
||||
elif action == "remove_file":
|
||||
if not file_path:
|
||||
return json.dumps({"success": False, "error": "file_path is required for 'remove_file'."}, ensure_ascii=False)
|
||||
return tool_error("file_path is required for 'remove_file'.", success=False)
|
||||
result = _remove_file(name, file_path)
|
||||
|
||||
else:
|
||||
@@ -727,7 +839,7 @@ SKILL_MANAGE_SCHEMA = {
|
||||
|
||||
|
||||
# --- Registry ---
|
||||
from tools.registry import registry
|
||||
from tools.registry import registry, tool_error
|
||||
|
||||
registry.register(
|
||||
name="skill_manage",
|
||||
|
||||
Reference in New Issue
Block a user