78 lines
2.9 KiB
Python
78 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
import shutil
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
from ..models import Patch
|
|
from ..storage import atomic_write_text, sha256_file
|
|
|
|
|
|
def _validate_skill_text(text: str) -> None:
|
|
if not text.strip():
|
|
raise ValueError("SKILL.md is empty")
|
|
if text.startswith("---"):
|
|
end = text.find("\n---", 3)
|
|
if end < 0:
|
|
raise ValueError("SKILL.md has an unterminated YAML frontmatter")
|
|
frontmatter = text[3:end].strip()
|
|
try:
|
|
import yaml
|
|
|
|
parsed = yaml.safe_load(frontmatter) if frontmatter else {}
|
|
except Exception as exc:
|
|
raise ValueError(f"invalid SKILL.md frontmatter: {exc}") from exc
|
|
if parsed is not None and not isinstance(parsed, dict):
|
|
raise ValueError("SKILL.md frontmatter must be a mapping")
|
|
|
|
|
|
def apply_patches(
|
|
current_package: Path, candidate_package: Path, patches: list[Patch]
|
|
) -> str:
|
|
skill = current_package / "SKILL.md"
|
|
if not skill.is_file():
|
|
raise ValueError(f"missing {skill}")
|
|
if not patches:
|
|
raise ValueError("at least one patch is required")
|
|
actual_hash = sha256_file(skill)
|
|
text = skill.read_text(encoding="utf-8")
|
|
for index, patch in enumerate(patches, start=1):
|
|
if not patch.skill_hash:
|
|
raise ValueError(f"patch {index} missing generation-time skill_hash")
|
|
if actual_hash != patch.skill_hash:
|
|
raise ValueError(
|
|
f"patch {index} was not generated from the current SKILL.md"
|
|
)
|
|
if not patch.old_text:
|
|
raise ValueError(f"patch {index} old_text must not be empty")
|
|
matches = text.count(patch.old_text)
|
|
if matches != 1:
|
|
raise ValueError(
|
|
f"patch {index} old_text must match exactly once; found {matches}"
|
|
)
|
|
changed = text.replace(patch.old_text, patch.new_text, 1)
|
|
if changed == text:
|
|
raise ValueError(f"patch {index} does not change SKILL.md")
|
|
_validate_skill_text(changed)
|
|
text = changed
|
|
token = uuid.uuid4().hex
|
|
temporary = candidate_package.with_name(f".{candidate_package.name}.{token}.tmp")
|
|
backup = candidate_package.with_name(f".{candidate_package.name}.{token}.bak")
|
|
shutil.copytree(current_package, temporary)
|
|
try:
|
|
atomic_write_text(temporary / "SKILL.md", text)
|
|
if candidate_package.exists():
|
|
candidate_package.rename(backup)
|
|
try:
|
|
temporary.rename(candidate_package)
|
|
except Exception:
|
|
if backup.exists() and not candidate_package.exists():
|
|
backup.rename(candidate_package)
|
|
raise
|
|
finally:
|
|
if temporary.exists():
|
|
shutil.rmtree(temporary)
|
|
if backup.exists() and candidate_package.exists():
|
|
shutil.rmtree(backup)
|
|
return sha256_file(candidate_package / "SKILL.md")
|