Compare commits
6 Commits
am/324-177
...
fix/624-er
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f12e1e69a7 | ||
| 3f525dd5a1 | |||
| a5902d5666 | |||
| 8657ea47ad | |||
| 954fd992eb | |||
|
|
f35f56e397 |
28
run_agent.py
28
run_agent.py
@@ -1001,30 +1001,10 @@ class AIAgent:
|
||||
self._session_db = session_db
|
||||
self._parent_session_id = parent_session_id
|
||||
self._last_flushed_db_idx = 0 # tracks DB-write cursor to prevent duplicate writes
|
||||
if self._session_db:
|
||||
try:
|
||||
self._session_db.create_session(
|
||||
session_id=self.session_id,
|
||||
source=self.platform or os.environ.get("HERMES_SESSION_SOURCE", "cli"),
|
||||
model=self.model,
|
||||
model_config={
|
||||
"max_iterations": self.max_iterations,
|
||||
"reasoning_config": reasoning_config,
|
||||
"max_tokens": max_tokens,
|
||||
},
|
||||
user_id=None,
|
||||
parent_session_id=self._parent_session_id,
|
||||
)
|
||||
except Exception as e:
|
||||
# Transient SQLite lock contention (e.g. CLI and gateway writing
|
||||
# concurrently) must NOT permanently disable session_search for
|
||||
# this agent. Keep _session_db alive — subsequent message
|
||||
# flushes and session_search calls will still work once the
|
||||
# lock clears. The session row may be missing from the index
|
||||
# for this run, but that is recoverable (flushes upsert rows).
|
||||
logger.warning(
|
||||
"Session DB create_session failed (session_search still available): %s", e
|
||||
)
|
||||
# Lazy session creation: defer until first message flush (#314).
|
||||
# _flush_messages_to_session_db() calls ensure_session() which uses
|
||||
# INSERT OR IGNORE — creating the row only when messages arrive.
|
||||
# This eliminates 32% of sessions that are created but never used.
|
||||
|
||||
# In-memory todo list for task planning (one per agent/session)
|
||||
from tools.todo_tool import TodoStore
|
||||
|
||||
152
tests/test_skill_manager_error_context.py
Normal file
152
tests/test_skill_manager_error_context.py
Normal file
@@ -0,0 +1,152 @@
|
||||
"""
|
||||
Tests for improved error messages in skill_manager_tool (issue #624).
|
||||
Verifies that error messages include file paths, context, and suggestions.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch, MagicMock
|
||||
from tools.skill_manager_tool import _format_error, _edit_skill, _patch_skill, skill_manage
|
||||
|
||||
|
||||
class TestFormatError:
|
||||
"""Test the _format_error helper function."""
|
||||
|
||||
def test_basic_error(self):
|
||||
"""Test basic error formatting."""
|
||||
result = _format_error("Something went wrong")
|
||||
assert result["success"] is False
|
||||
assert "Something went wrong" in result["error"]
|
||||
assert result["skill_name"] is None
|
||||
assert result["file_path"] is None
|
||||
|
||||
def test_with_skill_name(self):
|
||||
"""Test error with skill name."""
|
||||
result = _format_error("Failed", skill_name="test-skill")
|
||||
assert "test-skill" in result["error"]
|
||||
assert result["skill_name"] == "test-skill"
|
||||
|
||||
def test_with_file_path(self):
|
||||
"""Test error with file path."""
|
||||
result = _format_error("Failed", file_path="/path/to/SKILL.md")
|
||||
assert "/path/to/SKILL.md" in result["error"]
|
||||
assert result["file_path"] == "/path/to/SKILL.md"
|
||||
|
||||
def test_with_suggestion(self):
|
||||
"""Test error with suggestion."""
|
||||
result = _format_error("Failed", suggestion="Try again")
|
||||
assert "Suggestion: Try again" in result["error"]
|
||||
assert result["suggestion"] == "Try again"
|
||||
|
||||
def test_with_context(self):
|
||||
"""Test error with context dict."""
|
||||
result = _format_error("Failed", context={"line": 5, "found": "x"})
|
||||
assert "line: 5" in result["error"]
|
||||
assert "found: x" in result["error"]
|
||||
|
||||
def test_all_fields(self):
|
||||
"""Test error with all fields."""
|
||||
result = _format_error(
|
||||
"Pattern match failed",
|
||||
skill_name="my-skill",
|
||||
file_path="/skills/my-skill/SKILL.md",
|
||||
suggestion="Check whitespace",
|
||||
context={"expected": "foo", "found": "bar"}
|
||||
)
|
||||
assert "Pattern match failed" in result["error"]
|
||||
assert "Skill: my-skill" in result["error"]
|
||||
assert "File: /skills/my-skill/SKILL.md" in result["error"]
|
||||
assert "Suggestion: Check whitespace" in result["error"]
|
||||
assert "expected: foo" in result["error"]
|
||||
|
||||
|
||||
class TestEditSkillErrors:
|
||||
"""Test improved error messages in _edit_skill."""
|
||||
|
||||
@patch('tools.skill_manager_tool._find_skill')
|
||||
def test_skill_not_found(self, mock_find):
|
||||
"""Test skill not found error includes suggestion."""
|
||||
mock_find.return_value = None
|
||||
# Provide valid content with frontmatter so it passes validation
|
||||
valid_content = """---
|
||||
name: test
|
||||
description: Test skill
|
||||
---
|
||||
Body content here.
|
||||
"""
|
||||
result = _edit_skill("nonexistent", valid_content)
|
||||
assert result["success"] is False
|
||||
assert "nonexistent" in result["error"]
|
||||
assert "skills_list()" in result.get("suggestion", "")
|
||||
|
||||
@patch('tools.skill_manager_tool._find_skill')
|
||||
def test_yaml_parse_error_includes_file_path_and_line_number(self, mock_find, tmp_path):
|
||||
"""Invalid YAML should report target file path and parser line information."""
|
||||
skill_dir = tmp_path / "my-skill"
|
||||
skill_dir.mkdir()
|
||||
(skill_dir / "SKILL.md").write_text("old", encoding="utf-8")
|
||||
mock_find.return_value = {"path": skill_dir}
|
||||
|
||||
bad_content = """---
|
||||
name: my-skill
|
||||
description: [broken
|
||||
---
|
||||
Body.
|
||||
"""
|
||||
result = _edit_skill("my-skill", bad_content)
|
||||
assert result["success"] is False
|
||||
assert str(skill_dir / "SKILL.md") in result["error"]
|
||||
assert "line" in result["error"].lower()
|
||||
|
||||
|
||||
class TestPatchSkillErrors:
|
||||
"""Test improved error messages in _patch_skill."""
|
||||
|
||||
def test_old_string_required(self):
|
||||
"""Test old_string required error includes suggestion."""
|
||||
result = _patch_skill("test-skill", None, "new")
|
||||
assert result["success"] is False
|
||||
assert "old_string is required" in result["error"]
|
||||
assert "suggestion" in result
|
||||
|
||||
def test_new_string_required(self):
|
||||
"""Test new_string required error includes suggestion."""
|
||||
result = _patch_skill("test-skill", "old", None)
|
||||
assert result["success"] is False
|
||||
assert "new_string is required" in result["error"]
|
||||
assert "suggestion" in result
|
||||
|
||||
@patch('tools.skill_manager_tool._find_skill')
|
||||
def test_skill_not_found(self, mock_find):
|
||||
"""Test skill not found error includes suggestion."""
|
||||
mock_find.return_value = None
|
||||
result = _patch_skill("nonexistent", "old", "new")
|
||||
assert result["success"] is False
|
||||
assert "nonexistent" in result["error"]
|
||||
assert "skills_list()" in result.get("suggestion", "")
|
||||
|
||||
@patch('tools.skill_manager_tool._find_skill')
|
||||
def test_pattern_match_error_includes_state_info(self, mock_find, tmp_path):
|
||||
"""Patch failures should include file path and patch state info."""
|
||||
skill_dir = tmp_path / "state-skill"
|
||||
skill_dir.mkdir()
|
||||
target = skill_dir / "SKILL.md"
|
||||
target.write_text("---\nname: state-skill\ndescription: desc\n---\n\nBody content here.\n", encoding="utf-8")
|
||||
mock_find.return_value = {"path": skill_dir}
|
||||
|
||||
result = _patch_skill("state-skill", "missing pattern", "new text", replace_all=False)
|
||||
assert result["success"] is False
|
||||
assert str(target) in result["error"]
|
||||
assert "replace_all" in result["error"]
|
||||
assert "False" in result["error"]
|
||||
|
||||
|
||||
class TestSkillManageEntryPoint:
|
||||
def test_patch_missing_old_string_returns_json_error(self):
|
||||
result = skill_manage(action="patch", name="demo-skill", old_string="", new_string="x")
|
||||
assert isinstance(result, str)
|
||||
assert "old_string is required" in result
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
@@ -21,6 +21,18 @@ from typing import Callable, Dict, List, Optional, Set
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def tool_error(message: str, success: bool = False, **extra) -> str:
|
||||
"""Return a standardized JSON error payload for tool handlers.
|
||||
|
||||
Many tools import this helper directly from the registry module.
|
||||
Keeping it here avoids circular helper imports and ensures a consistent
|
||||
error envelope across tools.
|
||||
"""
|
||||
payload = {"success": success, "error": message}
|
||||
payload.update(extra)
|
||||
return json.dumps(payload, ensure_ascii=False)
|
||||
|
||||
|
||||
class ToolEntry:
|
||||
"""Metadata for a single registered tool."""
|
||||
|
||||
|
||||
@@ -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
|
||||
# =============================================================================
|
||||
@@ -140,43 +186,58 @@ def _validate_category(category: Optional[str]) -> Optional[str]:
|
||||
return None
|
||||
|
||||
|
||||
def _validate_frontmatter(content: str) -> Optional[str]:
|
||||
"""
|
||||
Validate that SKILL.md content has proper frontmatter with required fields.
|
||||
Returns error message or None if valid.
|
||||
def _validate_frontmatter_details(content: str) -> Tuple[Optional[str], Optional[dict]]:
|
||||
"""Validate SKILL.md frontmatter and return optional structured context.
|
||||
|
||||
Returns:
|
||||
(error_message, context_dict) where both are None when valid.
|
||||
"""
|
||||
if not content.strip():
|
||||
return "Content cannot be empty."
|
||||
return "Content cannot be empty.", None
|
||||
|
||||
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 (---). See existing skills for format.", None
|
||||
|
||||
end_match = re.search(r'\n---\s*\n', content[3:])
|
||||
if not end_match:
|
||||
return "SKILL.md frontmatter is not closed. Ensure you have a closing '---' line."
|
||||
return "SKILL.md frontmatter is not closed. Ensure you have a closing '---' line.", None
|
||||
|
||||
yaml_content = content[3:end_match.start() + 3]
|
||||
|
||||
try:
|
||||
parsed = yaml.safe_load(yaml_content)
|
||||
except yaml.YAMLError as e:
|
||||
return f"YAML frontmatter parse error: {e}"
|
||||
context = {}
|
||||
problem_mark = getattr(e, "problem_mark", None)
|
||||
if problem_mark is not None:
|
||||
context["line"] = problem_mark.line + 1
|
||||
context["column"] = problem_mark.column + 1
|
||||
return f"YAML frontmatter parse error: {e}", (context or None)
|
||||
|
||||
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.", None
|
||||
|
||||
if "name" not in parsed:
|
||||
return "Frontmatter must include 'name' field."
|
||||
return "Frontmatter must include 'name' field.", None
|
||||
if "description" not in parsed:
|
||||
return "Frontmatter must include 'description' field."
|
||||
return "Frontmatter must include 'description' field.", None
|
||||
if len(str(parsed["description"])) > MAX_DESCRIPTION_LENGTH:
|
||||
return f"Description exceeds {MAX_DESCRIPTION_LENGTH} characters."
|
||||
return f"Description exceeds {MAX_DESCRIPTION_LENGTH} characters.", None
|
||||
|
||||
body = content[end_match.end() + 3:].strip()
|
||||
if not body:
|
||||
return "SKILL.md must have content after the frontmatter (instructions, procedures, etc.)."
|
||||
return "SKILL.md must have content after the frontmatter (instructions, procedures, etc.).", None
|
||||
|
||||
return None
|
||||
return None, None
|
||||
|
||||
|
||||
def _validate_frontmatter(content: str) -> Optional[str]:
|
||||
"""
|
||||
Validate that SKILL.md content has proper frontmatter with required fields.
|
||||
Returns error message or None if valid.
|
||||
"""
|
||||
err, _context = _validate_frontmatter_details(content)
|
||||
return err
|
||||
|
||||
|
||||
def _validate_content_size(content: str, label: str = "SKILL.md") -> Optional[str]:
|
||||
@@ -210,7 +271,15 @@ def _find_skill(name: str) -> Optional[Dict[str, Any]]:
|
||||
{"path": Path} or None.
|
||||
"""
|
||||
from agent.skill_utils import get_all_skills_dirs
|
||||
for skills_dir in get_all_skills_dirs():
|
||||
|
||||
candidate_dirs = []
|
||||
if isinstance(SKILLS_DIR, Path):
|
||||
candidate_dirs.append(SKILLS_DIR)
|
||||
for extra_dir in get_all_skills_dirs():
|
||||
if extra_dir not in candidate_dirs:
|
||||
candidate_dirs.append(extra_dir)
|
||||
|
||||
for skills_dir in candidate_dirs:
|
||||
if not skills_dir.exists():
|
||||
continue
|
||||
for skill_md in skills_dir.rglob("SKILL.md"):
|
||||
@@ -224,13 +293,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 +316,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.
|
||||
@@ -292,10 +374,19 @@ def _create_skill(name: str, content: str, category: str = None) -> Dict[str, An
|
||||
if err:
|
||||
return {"success": False, "error": err}
|
||||
|
||||
skill_dir = _resolve_skill_dir(name, category)
|
||||
skill_md = skill_dir / "SKILL.md"
|
||||
|
||||
# Validate content
|
||||
err = _validate_frontmatter(content)
|
||||
err, context = _validate_frontmatter_details(content)
|
||||
if err:
|
||||
return {"success": False, "error": err}
|
||||
return _format_error(
|
||||
err,
|
||||
skill_name=name,
|
||||
file_path=str(skill_md),
|
||||
context=context,
|
||||
suggestion="Fix the YAML frontmatter before creating the skill."
|
||||
)
|
||||
|
||||
err = _validate_content_size(content)
|
||||
if err:
|
||||
@@ -304,24 +395,29 @@ 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)
|
||||
skill_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Write SKILL.md atomically
|
||||
skill_md = skill_dir / "SKILL.md"
|
||||
_atomic_write_text(skill_md, content)
|
||||
|
||||
# Security scan — roll back on block
|
||||
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,
|
||||
@@ -340,19 +436,30 @@ def _create_skill(name: str, content: str, category: str = None) -> Dict[str, An
|
||||
|
||||
def _edit_skill(name: str, content: str) -> Dict[str, Any]:
|
||||
"""Replace the SKILL.md of any existing skill (full rewrite)."""
|
||||
err = _validate_frontmatter(content)
|
||||
existing = _find_skill(name)
|
||||
if not existing:
|
||||
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"
|
||||
|
||||
err, context = _validate_frontmatter_details(content)
|
||||
if err:
|
||||
return {"success": False, "error": err}
|
||||
return _format_error(
|
||||
err,
|
||||
skill_name=name,
|
||||
file_path=str(skill_md),
|
||||
context=context,
|
||||
suggestion="Fix the YAML frontmatter before updating the skill."
|
||||
)
|
||||
|
||||
err = _validate_content_size(content)
|
||||
if err:
|
||||
return {"success": False, "error": err}
|
||||
|
||||
existing = _find_skill(name)
|
||||
if not existing:
|
||||
return {"success": False, "error": f"Skill '{name}' not found. Use skills_list() to see available skills."}
|
||||
|
||||
skill_md = existing["path"] / "SKILL.md"
|
||||
# Back up original content for rollback
|
||||
original_content = skill_md.read_text(encoding="utf-8") if skill_md.exists() else None
|
||||
_atomic_write_text(skill_md, content)
|
||||
@@ -362,7 +469,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 +496,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 +521,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")
|
||||
|
||||
@@ -421,11 +550,18 @@ def _patch_skill(
|
||||
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={
|
||||
"replace_all": replace_all,
|
||||
"target_exists": target.exists(),
|
||||
"content_chars": len(content),
|
||||
"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
|
||||
@@ -435,12 +571,15 @@ def _patch_skill(
|
||||
|
||||
# If patching SKILL.md, validate frontmatter is still intact
|
||||
if not file_path:
|
||||
err = _validate_frontmatter(new_content)
|
||||
err, validation_context = _validate_frontmatter_details(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),
|
||||
context=validation_context,
|
||||
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 +600,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 +646,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 +684,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 +744,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 +764,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 +882,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