Compare commits
2 Commits
fix/format
...
fix/923
| Author | SHA1 | Date | |
|---|---|---|---|
| d27ca6d39a | |||
| c6f2855745 |
122
tools/skill_edit_guard.py
Normal file
122
tools/skill_edit_guard.py
Normal file
@@ -0,0 +1,122 @@
|
|||||||
|
"""Skill Edit Guard — Poka-yoke auto-revert for incomplete skill edits.
|
||||||
|
|
||||||
|
Creates atomic skill edits with automatic rollback on failure.
|
||||||
|
Prevents broken skills from corrupting future sessions.
|
||||||
|
|
||||||
|
Usage:
|
||||||
|
from tools.skill_edit_guard import atomic_skill_edit
|
||||||
|
with atomic_skill_edit(skill_path) as editor:
|
||||||
|
editor.write(new_content)
|
||||||
|
# If exception occurs, file is automatically reverted
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
|
import os
|
||||||
|
import shutil
|
||||||
|
import tempfile
|
||||||
|
import time
|
||||||
|
from contextlib import contextmanager
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import Any, Dict, Optional
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
class SkillEditGuard:
|
||||||
|
"""Atomic skill file editing with auto-revert on failure."""
|
||||||
|
|
||||||
|
def __init__(self, skill_path: str):
|
||||||
|
self._path = Path(skill_path)
|
||||||
|
self._backup: Optional[Path] = None
|
||||||
|
self._committed = False
|
||||||
|
|
||||||
|
def backup(self) -> bool:
|
||||||
|
"""Create backup before editing."""
|
||||||
|
if not self._path.exists():
|
||||||
|
return True # New file, nothing to backup
|
||||||
|
|
||||||
|
backup_dir = self._path.parent / ".skill_backups"
|
||||||
|
backup_dir.mkdir(exist_ok=True)
|
||||||
|
|
||||||
|
ts = int(time.time() * 1000)
|
||||||
|
self._backup = backup_dir / f"{self._path.name}.{ts}.bak"
|
||||||
|
shutil.copy2(self._path, self._backup)
|
||||||
|
logger.debug("Skill backup created: %s", self._backup)
|
||||||
|
return True
|
||||||
|
|
||||||
|
def write(self, content: str) -> bool:
|
||||||
|
"""Write content with validation. Returns True if valid."""
|
||||||
|
# Validate YAML frontmatter
|
||||||
|
if content.startswith("---"):
|
||||||
|
end = content.find("---", 3)
|
||||||
|
if end < 0:
|
||||||
|
logger.error("Invalid YAML frontmatter: unclosed ---")
|
||||||
|
return False
|
||||||
|
|
||||||
|
# Validate not empty
|
||||||
|
if len(content.strip()) < 10:
|
||||||
|
logger.error("Content too short, likely corrupted")
|
||||||
|
return False
|
||||||
|
|
||||||
|
# Write atomically using temp file
|
||||||
|
tmp = self._path.with_suffix(".tmp")
|
||||||
|
try:
|
||||||
|
tmp.write_text(content, encoding="utf-8")
|
||||||
|
tmp.rename(self._path)
|
||||||
|
return True
|
||||||
|
except Exception as e:
|
||||||
|
logger.error("Write failed: %s", e)
|
||||||
|
if tmp.exists():
|
||||||
|
tmp.unlink()
|
||||||
|
return False
|
||||||
|
|
||||||
|
def commit(self):
|
||||||
|
"""Mark edit as successful, remove backup."""
|
||||||
|
self._committed = True
|
||||||
|
if self._backup and self._backup.exists():
|
||||||
|
self._backup.unlink()
|
||||||
|
logger.debug("Skill backup removed: %s", self._backup)
|
||||||
|
|
||||||
|
def rollback(self) -> bool:
|
||||||
|
"""Revert to backup."""
|
||||||
|
if self._backup and self._backup.exists():
|
||||||
|
shutil.copy2(self._backup, self._path)
|
||||||
|
self._backup.unlink()
|
||||||
|
logger.warning("Skill reverted from backup: %s", self._path)
|
||||||
|
return True
|
||||||
|
return False
|
||||||
|
|
||||||
|
def __enter__(self):
|
||||||
|
self.backup()
|
||||||
|
return self
|
||||||
|
|
||||||
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||||
|
if exc_type is not None:
|
||||||
|
self.rollback()
|
||||||
|
return False # Re-raise exception
|
||||||
|
if not self._committed:
|
||||||
|
self.rollback()
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
@contextmanager
|
||||||
|
def atomic_skill_edit(skill_path: str):
|
||||||
|
"""Context manager for atomic skill editing.
|
||||||
|
|
||||||
|
Usage:
|
||||||
|
with atomic_skill_edit("/path/to/skill/SKILL.md") as editor:
|
||||||
|
success = editor.write(new_content)
|
||||||
|
if not success:
|
||||||
|
raise ValueError("Write failed")
|
||||||
|
# __exit__ commits on success, reverts on exception
|
||||||
|
"""
|
||||||
|
guard = SkillEditGuard(skill_path)
|
||||||
|
guard.backup()
|
||||||
|
try:
|
||||||
|
yield guard
|
||||||
|
guard.commit()
|
||||||
|
except Exception:
|
||||||
|
guard.rollback()
|
||||||
|
raise
|
||||||
@@ -44,6 +44,34 @@ from typing import Dict, Any, Optional, Tuple
|
|||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
def _format_error(
|
||||||
|
message: str,
|
||||||
|
skill_name: str = None,
|
||||||
|
file_path: str = None,
|
||||||
|
suggestion: str = None,
|
||||||
|
context: dict = None,
|
||||||
|
) -> Dict[str, Any]:
|
||||||
|
"""Format an error with rich context for better debugging."""
|
||||||
|
parts = [message]
|
||||||
|
if skill_name:
|
||||||
|
parts.append(f"Skill: {skill_name}")
|
||||||
|
if file_path:
|
||||||
|
parts.append(f"File: {file_path}")
|
||||||
|
if suggestion:
|
||||||
|
parts.append(f"Suggestion: {suggestion}")
|
||||||
|
if context:
|
||||||
|
for key, value in context.items():
|
||||||
|
parts.append(f"{key}: {value}")
|
||||||
|
return {
|
||||||
|
"success": False,
|
||||||
|
"error": " | ".join(parts),
|
||||||
|
"skill_name": skill_name,
|
||||||
|
"file_path": file_path,
|
||||||
|
"suggestion": suggestion,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
# Import security scanner — agent-created skills get the same scrutiny as
|
# Import security scanner — agent-created skills get the same scrutiny as
|
||||||
# community hub installs.
|
# community hub installs.
|
||||||
try:
|
try:
|
||||||
|
|||||||
Reference in New Issue
Block a user