Compare commits
4 Commits
whip/251-1
...
fix/624-er
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f12e1e69a7 | ||
| 3f525dd5a1 | |||
| a5902d5666 | |||
| 8657ea47ad |
@@ -234,7 +234,12 @@ class HolographicMemoryProvider(MemoryProvider):
|
||||
return self._handle_fact_feedback(args)
|
||||
return json.dumps({"error": f"Unknown tool: {tool_name}"})
|
||||
|
||||
|
||||
def on_session_end(self, messages: List[Dict[str, Any]]) -> None:
|
||||
if not self._config.get("auto_extract", False):
|
||||
return
|
||||
if not self._store or not messages:
|
||||
return
|
||||
self._auto_extract_facts(messages)
|
||||
|
||||
def on_memory_write(self, action: str, target: str, content: str) -> None:
|
||||
"""Mirror built-in memory writes as facts.
|
||||
@@ -261,44 +266,6 @@ class HolographicMemoryProvider(MemoryProvider):
|
||||
except Exception as e:
|
||||
logger.debug("Holographic memory_write mirror failed: %s", e)
|
||||
|
||||
def on_session_end(self, messages: List[Dict[str, Any]]) -> None:
|
||||
"""Run auto-extraction and periodic contradiction detection."""
|
||||
if self._config.get("auto_extract", False):
|
||||
self._auto_extract_facts(messages)
|
||||
|
||||
# Periodic contradiction detection (run every ~50 sessions or on first session)
|
||||
self._maybe_run_contradiction_scan()
|
||||
|
||||
def _maybe_run_contradiction_scan(self) -> None:
|
||||
"""Run contradiction detection if enough sessions have passed since last run."""
|
||||
if not self._store or not self._retriever:
|
||||
return
|
||||
try:
|
||||
# Use a counter file to track sessions since last scan
|
||||
from hermes_constants import get_hermes_home
|
||||
counter_path = get_hermes_home() / ".contradiction_scan_counter"
|
||||
|
||||
count = 0
|
||||
if counter_path.exists():
|
||||
try:
|
||||
count = int(counter_path.read_text().strip())
|
||||
except (ValueError, OSError):
|
||||
count = 0
|
||||
|
||||
count += 1
|
||||
counter_path.write_text(str(count))
|
||||
|
||||
# Run every 50 sessions
|
||||
if count >= 50:
|
||||
counter_path.write_text("0")
|
||||
from .resolver import ContradictionResolver
|
||||
resolver = ContradictionResolver(self._store, self._retriever)
|
||||
report = resolver.run_detection_and_resolution(limit=50, auto_resolve=True)
|
||||
if report.contradictions_found > 0:
|
||||
logger.info("Periodic contradiction scan: %s", report.summary())
|
||||
except Exception as e:
|
||||
logger.debug("Periodic contradiction scan failed: %s", e)
|
||||
|
||||
def shutdown(self) -> None:
|
||||
self._store = None
|
||||
self._retriever = None
|
||||
|
||||
@@ -1,179 +0,0 @@
|
||||
"""Contradiction detection and resolution for holographic memory.
|
||||
|
||||
Periodically scans the fact store for contradictions using the retriever's
|
||||
contradict() method, then resolves obvious cases and flags ambiguous ones.
|
||||
|
||||
Resolution strategy:
|
||||
- Obvious: same entity, newer fact supersedes older → lower trust on older
|
||||
- Ambiguous: different claims about same entity → flag for review, don't auto-resolve
|
||||
- High-confidence contradiction (>0.7 score): lower trust on both, log warning
|
||||
|
||||
Usage:
|
||||
from plugins.memory.holographic.resolver import ContradictionResolver
|
||||
resolver = ContradictionResolver(store, retriever)
|
||||
report = resolver.run_detection_and_resolution()
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Thresholds
|
||||
_OBVIOUS_THRESHOLD = 0.5 # contradiction_score >= this → obvious
|
||||
_AMBIGUOUS_THRESHOLD = 0.3 # contradiction_score >= this → flag
|
||||
_HIGH_CONFIDENCE = 0.7 # contradiction_score >= this → high confidence
|
||||
_TRUST_PENALTY_OLD = 0.3 # trust reduction for superseded fact
|
||||
_TRUST_PENALTY_CONFLICT = 0.15 # trust reduction for ambiguous conflicts
|
||||
|
||||
|
||||
class ContradictionReport:
|
||||
"""Results of a contradiction detection and resolution run."""
|
||||
|
||||
def __init__(self):
|
||||
self.total_scanned: int = 0
|
||||
self.contradictions_found: int = 0
|
||||
self.auto_resolved: int = 0
|
||||
self.flagged_for_review: int = 0
|
||||
self.high_confidence: int = 0
|
||||
self.resolved_pairs: List[Dict[str, Any]] = []
|
||||
self.flagged_pairs: List[Dict[str, Any]] = []
|
||||
|
||||
def summary(self) -> str:
|
||||
lines = [
|
||||
f"Contradiction scan: {self.total_scanned} facts, "
|
||||
f"{self.contradictions_found} contradictions found",
|
||||
f" Auto-resolved: {self.auto_resolved} (newer supersedes older)",
|
||||
f" High-confidence: {self.high_confidence} (trust lowered on both)",
|
||||
f" Flagged for review: {self.flagged_for_review}",
|
||||
]
|
||||
for pair in self.flagged_pairs[:5]:
|
||||
lines.append(
|
||||
f" [REVIEW] score={pair['contradiction_score']:.2f} "
|
||||
f"entities={pair['shared_entities']} "
|
||||
f"A: {pair['fact_a']['content'][:50]}… "
|
||||
f"B: {pair['fact_b']['content'][:50]}…"
|
||||
)
|
||||
return "\n".join(lines)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"total_scanned": self.total_scanned,
|
||||
"contradictions_found": self.contradictions_found,
|
||||
"auto_resolved": self.auto_resolved,
|
||||
"high_confidence": self.high_confidence,
|
||||
"flagged_for_review": self.flagged_for_review,
|
||||
"resolved_pairs": self.resolved_pairs,
|
||||
"flagged_pairs": self.flagged_pairs,
|
||||
}
|
||||
|
||||
|
||||
class ContradictionResolver:
|
||||
"""Detects and resolves contradictions in the holographic fact store."""
|
||||
|
||||
def __init__(self, store, retriever):
|
||||
self._store = store
|
||||
self._retriever = retriever
|
||||
|
||||
def run_detection_and_resolution(
|
||||
self,
|
||||
limit: int = 50,
|
||||
auto_resolve: bool = True,
|
||||
) -> ContradictionReport:
|
||||
"""Run a full contradiction detection and resolution pass.
|
||||
|
||||
Args:
|
||||
limit: Max contradiction pairs to process.
|
||||
auto_resolve: If True, auto-resolve obvious cases.
|
||||
|
||||
Returns:
|
||||
ContradictionReport with all findings and actions taken.
|
||||
"""
|
||||
report = ContradictionReport()
|
||||
|
||||
try:
|
||||
contradictions = self._retriever.contradict(limit=limit)
|
||||
except Exception as e:
|
||||
logger.warning("Contradiction detection failed: %s", e)
|
||||
return report
|
||||
|
||||
report.total_scanned = len(contradictions)
|
||||
report.contradictions_found = len(contradictions)
|
||||
|
||||
for pair in contradictions:
|
||||
score = pair.get("contradiction_score", 0.0)
|
||||
|
||||
if score >= _HIGH_CONFIDENCE:
|
||||
report.high_confidence += 1
|
||||
if auto_resolve:
|
||||
self._resolve_high_confidence(pair, report)
|
||||
elif score >= _OBVIOUS_THRESHOLD:
|
||||
if auto_resolve:
|
||||
self._resolve_obvious(pair, report)
|
||||
elif score >= _AMBIGUOUS_THRESHOLD:
|
||||
report.flagged_for_review += 1
|
||||
report.flagged_pairs.append(pair)
|
||||
# Lower trust slightly on both to reduce retrieval weight
|
||||
self._penalize_both(pair, _TRUST_PENALTY_CONFLICT)
|
||||
|
||||
return report
|
||||
|
||||
def _resolve_obvious(self, pair: dict, report: ContradictionReport) -> None:
|
||||
"""Resolve obvious contradiction: newer fact supersedes older.
|
||||
|
||||
Same entity, clearly contradictory claims. Newer wins.
|
||||
"""
|
||||
fact_a = pair["fact_a"]
|
||||
fact_b = pair["fact_b"]
|
||||
|
||||
# Determine which is newer
|
||||
a_time = fact_a.get("updated_at") or fact_a.get("created_at", "")
|
||||
b_time = fact_b.get("updated_at") or fact_b.get("created_at", "")
|
||||
|
||||
if a_time >= b_time:
|
||||
newer, older = fact_a, fact_b
|
||||
else:
|
||||
newer, older = fact_b, fact_a
|
||||
|
||||
# Lower trust on the older (superseded) fact
|
||||
try:
|
||||
self._store.update_fact(
|
||||
older["fact_id"],
|
||||
trust_delta=-_TRUST_PENALTY_OLD,
|
||||
)
|
||||
report.auto_resolved += 1
|
||||
report.resolved_pairs.append({
|
||||
"action": "superseded",
|
||||
"kept": newer["fact_id"],
|
||||
"demoted": older["fact_id"],
|
||||
"reason": f"Newer fact supersedes older (score={pair['contradiction_score']:.2f})",
|
||||
})
|
||||
logger.info(
|
||||
"Contradiction resolved: fact#%d supersedes fact#%d (entities: %s)",
|
||||
newer["fact_id"], older["fact_id"],
|
||||
pair.get("shared_entities", []),
|
||||
)
|
||||
except Exception as e:
|
||||
logger.debug("Failed to resolve contradiction: %s", e)
|
||||
|
||||
def _resolve_high_confidence(self, pair: dict, report: ContradictionReport) -> None:
|
||||
"""Resolve high-confidence contradiction: lower trust on both facts.
|
||||
|
||||
We can't determine which is correct, so both get penalized.
|
||||
"""
|
||||
self._penalize_both(pair, _TRUST_PENALTY_CONFLICT * 2)
|
||||
report.flagged_pairs.append(pair)
|
||||
|
||||
def _penalize_both(self, pair: dict, penalty: float) -> None:
|
||||
"""Lower trust on both contradictory facts."""
|
||||
for key in ("fact_a", "fact_b"):
|
||||
fact = pair.get(key, {})
|
||||
fid = fact.get("fact_id")
|
||||
if fid:
|
||||
try:
|
||||
self._store.update_fact(fid, trust_delta=-penalty)
|
||||
except Exception as e:
|
||||
logger.debug("Failed to penalize fact#%d: %s", fid, e)
|
||||
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