398 lines
13 KiB
Python
398 lines
13 KiB
Python
"""Filesystem-backed version history for custom skills.
|
|
|
|
Only the active package lives below ``skills/custom``. Historical snapshots
|
|
are kept under the writable DeerFlow runtime directory, outside the canonical
|
|
skills tree and therefore outside every model-facing discovery/mount path.
|
|
|
|
The feature is deliberately gated by ``system_settings.skill_versioning``.
|
|
When the switch is off, callers keep the legacy write behaviour and this
|
|
module performs no snapshot writes.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
import threading
|
|
from collections.abc import Iterator
|
|
from contextlib import contextmanager
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
from deerflow.config.paths import get_paths
|
|
from deerflow.config.system_settings import load_system_settings
|
|
from deerflow.skills.storage import SkillStorage, get_or_new_skill_storage
|
|
from deerflow.skills.types import SKILL_MD_FILE, SkillCategory
|
|
|
|
_MANIFEST_FILE = "manifest.json"
|
|
_SCHEMA_VERSION = 1
|
|
_LOCKS_GUARD = threading.Lock()
|
|
_THREAD_LOCKS: dict[str, threading.RLock] = {}
|
|
|
|
|
|
def skill_versioning_enabled() -> bool:
|
|
"""Return the deployment-wide custom-skill versioning switch."""
|
|
|
|
try:
|
|
return bool(load_system_settings().skill_versioning.enabled)
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _thread_lock(name: str) -> threading.RLock:
|
|
with _LOCKS_GUARD:
|
|
return _THREAD_LOCKS.setdefault(name, threading.RLock())
|
|
|
|
|
|
def _versions_root() -> Path:
|
|
return get_paths().base_dir / "skill-versions"
|
|
|
|
|
|
def _skill_root(name: str) -> Path:
|
|
return _versions_root() / SkillStorage.validate_skill_name(name)
|
|
|
|
|
|
def _manifest_path(name: str) -> Path:
|
|
return _skill_root(name) / _MANIFEST_FILE
|
|
|
|
|
|
def _version_dir(name: str, version: int) -> Path:
|
|
return _skill_root(name) / f"v{version:06d}"
|
|
|
|
|
|
@contextmanager
|
|
def _cross_process_lock(name: str) -> Iterator[None]:
|
|
"""Serialize one skill's snapshot/switch operations across workers."""
|
|
|
|
root = _skill_root(name)
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
lock_path = root / ".lock"
|
|
with _thread_lock(name):
|
|
with lock_path.open("a+b") as handle:
|
|
if os.name == "nt": # pragma: no cover - exercised on Windows hosts
|
|
import msvcrt
|
|
|
|
if lock_path.stat().st_size == 0:
|
|
handle.write(b"0")
|
|
handle.flush()
|
|
handle.seek(0)
|
|
msvcrt.locking(handle.fileno(), msvcrt.LK_LOCK, 1)
|
|
else: # pragma: no cover - exercised in Linux CI/deployments
|
|
import fcntl
|
|
|
|
fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
|
|
try:
|
|
yield
|
|
finally:
|
|
if os.name == "nt": # pragma: no cover
|
|
handle.seek(0)
|
|
msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
|
|
else: # pragma: no cover
|
|
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
|
|
|
|
|
|
def _empty_manifest(name: str) -> dict[str, Any]:
|
|
return {
|
|
"schema_version": _SCHEMA_VERSION,
|
|
"skill_name": name,
|
|
"active_version": None,
|
|
"versions": [],
|
|
}
|
|
|
|
|
|
def _load_manifest(name: str) -> dict[str, Any]:
|
|
path = _manifest_path(name)
|
|
if not path.is_file():
|
|
return _empty_manifest(name)
|
|
try:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
raise ValueError(f"Skill version manifest for '{name}' is unreadable") from exc
|
|
if not isinstance(payload, dict) or payload.get("skill_name") != name:
|
|
raise ValueError(f"Skill version manifest for '{name}' is invalid")
|
|
versions = payload.get("versions")
|
|
if not isinstance(versions, list):
|
|
raise ValueError(f"Skill version manifest for '{name}' has invalid versions")
|
|
return payload
|
|
|
|
|
|
def _save_manifest(name: str, manifest: dict[str, Any]) -> None:
|
|
path = _manifest_path(name)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
fd, tmp_name = tempfile.mkstemp(prefix=".manifest-", suffix=".tmp", dir=path.parent)
|
|
try:
|
|
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
|
json.dump(manifest, handle, ensure_ascii=False, indent=2, sort_keys=True)
|
|
handle.write("\n")
|
|
os.replace(tmp_name, path)
|
|
except Exception:
|
|
try:
|
|
os.unlink(tmp_name)
|
|
except OSError:
|
|
pass
|
|
raise
|
|
|
|
|
|
def _resolve_custom_dir(storage: SkillStorage, name: str) -> Path:
|
|
for skill in storage.load_skills(enabled_only=False):
|
|
if skill.name == name and skill.category == SkillCategory.CUSTOM:
|
|
return skill.skill_dir.resolve()
|
|
raise FileNotFoundError(f"Custom skill '{name}' not found")
|
|
|
|
|
|
def _tree_hash(root: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
for path in sorted(root.rglob("*"), key=lambda item: item.relative_to(root).as_posix()):
|
|
relative = path.relative_to(root).as_posix()
|
|
if any(part in {"__pycache__", ".git"} for part in path.relative_to(root).parts):
|
|
continue
|
|
if path.is_symlink():
|
|
digest.update(b"L\0")
|
|
digest.update(relative.encode("utf-8"))
|
|
digest.update(b"\0")
|
|
digest.update(os.readlink(path).encode("utf-8", errors="surrogateescape"))
|
|
continue
|
|
if not path.is_file():
|
|
continue
|
|
digest.update(b"F\0")
|
|
digest.update(relative.encode("utf-8"))
|
|
digest.update(b"\0")
|
|
with path.open("rb") as handle:
|
|
while chunk := handle.read(1024 * 1024):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def _copy_package(source: Path, destination: Path) -> None:
|
|
shutil.copytree(
|
|
source,
|
|
destination,
|
|
symlinks=True,
|
|
ignore=shutil.ignore_patterns("__pycache__", ".git"),
|
|
)
|
|
|
|
|
|
def record_custom_skill_version(
|
|
name: str,
|
|
*,
|
|
storage: SkillStorage | None = None,
|
|
action: str,
|
|
author: str | None = None,
|
|
summary: str | None = None,
|
|
) -> dict[str, Any] | None:
|
|
"""Snapshot the current custom package when it differs from active.
|
|
|
|
Returns the active/new version metadata. No-op (and no filesystem writes)
|
|
while the deployment switch is disabled.
|
|
"""
|
|
|
|
if not skill_versioning_enabled():
|
|
return None
|
|
name = SkillStorage.validate_skill_name(name)
|
|
storage = storage or get_or_new_skill_storage()
|
|
source = _resolve_custom_dir(storage, name)
|
|
current_hash = _tree_hash(source)
|
|
|
|
with _cross_process_lock(name):
|
|
manifest = _load_manifest(name)
|
|
active = next(
|
|
(item for item in manifest["versions"] if int(item.get("version") or 0) == int(manifest.get("active_version") or 0)),
|
|
None,
|
|
)
|
|
if active is not None and active.get("sha256") == current_hash:
|
|
return {**active, "active": True}
|
|
|
|
next_version = max([int(item.get("version") or 0) for item in manifest["versions"]] or [0]) + 1
|
|
destination = _version_dir(name, next_version)
|
|
staging = destination.parent / f".{destination.name}.staging-{uuid4().hex}"
|
|
_copy_package(source, staging)
|
|
try:
|
|
os.replace(staging, destination)
|
|
except Exception:
|
|
shutil.rmtree(staging, ignore_errors=True)
|
|
raise
|
|
|
|
now = datetime.now(UTC).isoformat()
|
|
entry = {
|
|
"version": next_version,
|
|
"sha256": current_hash,
|
|
"created_at": now,
|
|
"created_by": author,
|
|
"action": str(action or "update"),
|
|
"summary": str(summary or "").strip() or None,
|
|
"based_on_version": manifest.get("active_version"),
|
|
}
|
|
manifest["versions"].append(entry)
|
|
manifest["active_version"] = next_version
|
|
manifest["updated_at"] = now
|
|
_save_manifest(name, manifest)
|
|
return {**entry, "active": True}
|
|
|
|
|
|
def list_custom_skill_versions(
|
|
name: str,
|
|
*,
|
|
storage: SkillStorage | None = None,
|
|
author: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Return newest-first metadata, lazily bootstrapping the current package."""
|
|
|
|
if not skill_versioning_enabled():
|
|
return {"enabled": False, "skill_name": name, "active_version": None, "versions": []}
|
|
storage = storage or get_or_new_skill_storage()
|
|
record_custom_skill_version(
|
|
name,
|
|
storage=storage,
|
|
action="baseline",
|
|
author=author,
|
|
summary="启用技能迭代时保存的当前版本",
|
|
)
|
|
with _cross_process_lock(name):
|
|
manifest = _load_manifest(name)
|
|
active_version = manifest.get("active_version")
|
|
versions = [{**item, "active": int(item.get("version") or 0) == int(active_version or 0)} for item in reversed(manifest["versions"])]
|
|
return {
|
|
"enabled": True,
|
|
"skill_name": name,
|
|
"active_version": active_version,
|
|
"versions": versions,
|
|
}
|
|
|
|
|
|
def activate_custom_skill_version(
|
|
name: str,
|
|
version: int,
|
|
*,
|
|
storage: SkillStorage | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Materialize one historical snapshot as the sole model-visible package."""
|
|
|
|
if not skill_versioning_enabled():
|
|
raise RuntimeError("技能迭代功能未开启")
|
|
name = SkillStorage.validate_skill_name(name)
|
|
storage = storage or get_or_new_skill_storage()
|
|
target = _resolve_custom_dir(storage, name)
|
|
|
|
with _cross_process_lock(name):
|
|
manifest = _load_manifest(name)
|
|
entry = next(
|
|
(item for item in manifest["versions"] if int(item.get("version") or 0) == int(version)),
|
|
None,
|
|
)
|
|
if entry is None:
|
|
raise FileNotFoundError(f"Skill '{name}' version {version} not found")
|
|
source = _version_dir(name, int(version))
|
|
if not source.is_dir() or _tree_hash(source) != entry.get("sha256"):
|
|
raise ValueError(f"Skill '{name}' version {version} snapshot is missing or corrupted")
|
|
storage.validate_skill_markdown_content(
|
|
name,
|
|
(source / SKILL_MD_FILE).read_text(encoding="utf-8"),
|
|
)
|
|
if int(manifest.get("active_version") or 0) == int(version) and _tree_hash(target) == entry.get("sha256"):
|
|
return {**entry, "active": True}
|
|
|
|
parent = target.parent
|
|
staging = parent / f".{name}.version-stage-{uuid4().hex}"
|
|
backup = parent / f".{name}.version-backup-{uuid4().hex}"
|
|
_copy_package(source, staging)
|
|
moved_old = False
|
|
try:
|
|
os.replace(target, backup)
|
|
moved_old = True
|
|
os.replace(staging, target)
|
|
manifest["active_version"] = int(version)
|
|
manifest["updated_at"] = datetime.now(UTC).isoformat()
|
|
_save_manifest(name, manifest)
|
|
except Exception:
|
|
shutil.rmtree(staging, ignore_errors=True)
|
|
if moved_old:
|
|
shutil.rmtree(target, ignore_errors=True)
|
|
os.replace(backup, target)
|
|
raise
|
|
finally:
|
|
if backup.exists() and target.exists():
|
|
shutil.rmtree(backup, ignore_errors=True)
|
|
return {**entry, "active": True}
|
|
|
|
|
|
def delete_custom_skill_versions(name: str) -> None:
|
|
"""Remove stored snapshots after the custom skill itself is deleted."""
|
|
|
|
name = SkillStorage.validate_skill_name(name)
|
|
root = _skill_root(name)
|
|
if root.exists():
|
|
shutil.rmtree(root)
|
|
|
|
|
|
def reset_custom_skill_versions(
|
|
name: str,
|
|
*,
|
|
storage: SkillStorage | None = None,
|
|
author: str | None = None,
|
|
summary: str | None = None,
|
|
) -> dict[str, Any] | None:
|
|
"""Discard the previous chain and make the current package the new v1.
|
|
|
|
This is the explicit "direct overwrite" semantic offered beside
|
|
"iterate as a new version" for same-name uploads.
|
|
"""
|
|
|
|
if not skill_versioning_enabled():
|
|
return None
|
|
name = SkillStorage.validate_skill_name(name)
|
|
root = _skill_root(name)
|
|
with _cross_process_lock(name):
|
|
for child in list(root.iterdir()):
|
|
if child.name == ".lock":
|
|
continue
|
|
if child.is_dir():
|
|
shutil.rmtree(child)
|
|
else:
|
|
child.unlink(missing_ok=True)
|
|
return record_custom_skill_version(
|
|
name,
|
|
storage=storage,
|
|
action="overwrite",
|
|
author=author,
|
|
summary=summary or "直接覆盖后重置版本链",
|
|
)
|
|
|
|
|
|
def reconcile_custom_skill_versions(*, storage: SkillStorage | None = None, author: str | None = None) -> int:
|
|
"""Capture each current custom skill after the master switch is enabled."""
|
|
|
|
if not skill_versioning_enabled():
|
|
return 0
|
|
storage = storage or get_or_new_skill_storage()
|
|
captured = 0
|
|
for skill in storage.load_skills(enabled_only=False):
|
|
if skill.category != SkillCategory.CUSTOM:
|
|
continue
|
|
before = _load_manifest(skill.name).get("active_version") if _manifest_path(skill.name).exists() else None
|
|
result = record_custom_skill_version(
|
|
skill.name,
|
|
storage=storage,
|
|
action="feature_enabled",
|
|
author=author,
|
|
summary="技能迭代开关开启后捕获当前内容",
|
|
)
|
|
if result is not None and result.get("version") != before:
|
|
captured += 1
|
|
return captured
|
|
|
|
|
|
__all__ = [
|
|
"activate_custom_skill_version",
|
|
"delete_custom_skill_versions",
|
|
"list_custom_skill_versions",
|
|
"record_custom_skill_version",
|
|
"reconcile_custom_skill_versions",
|
|
"reset_custom_skill_versions",
|
|
"skill_versioning_enabled",
|
|
]
|