"""Skill usage sidecar — per-skill telemetry stored in .usage.json.""" from __future__ import annotations import json import logging import os import tempfile import threading from datetime import UTC, datetime from pathlib import Path logger = logging.getLogger(__name__) _USAGE_FILENAME = ".usage.json" # 串行化对 .usage.json 的"读-改-写",避免并发(尤其 curator 多线程)丢更新。 # .usage.json 是全局单文件;此锁覆盖进程内并发。leader 选举已使 curator 只在 # 单进程跑,故进程内锁即可保证状态流转不丢。 _usage_lock = threading.Lock() STATE_ACTIVE = "active" STATE_STALE = "stale" STATE_ARCHIVED = "archived" SOURCE_AGENT = "agent" SOURCE_UPLOAD = "upload" _DEFAULT_ENTRY: dict = { "use_count": 0, "view_count": 0, "patch_count": 0, "last_used_at": None, "last_viewed_at": None, "last_patched_at": None, "created_at": None, "state": STATE_ACTIVE, "pinned": False, "featured": False, "archived_at": None, "source": SOURCE_UPLOAD, } def _usage_path() -> Path: from deerflow.skills.storage import get_or_new_skill_storage return Path(get_or_new_skill_storage().get_skills_root_path()) / _USAGE_FILENAME def _read_all() -> dict: path = _usage_path() if not path.exists(): return {} try: return json.loads(path.read_text(encoding="utf-8")) except Exception: logger.debug("Failed to read %s", path, exc_info=True) return {} def _write_all(data: dict) -> None: path = _usage_path() try: path.parent.mkdir(parents=True, exist_ok=True) tmp_fd, tmp_path = tempfile.mkstemp(dir=path.parent, suffix=".tmp") try: with os.fdopen(tmp_fd, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, default=str) os.replace(tmp_path, path) except Exception: try: os.unlink(tmp_path) except Exception: pass raise except Exception: logger.debug("Failed to write %s", path, exc_info=True) def _now_iso() -> str: return datetime.now(UTC).isoformat() def _update_entry(name: str, updates) -> None: """更新一个 usage 条目。``updates`` 可以是 dict,或 ``(entry) -> dict`` 回调。 用回调形式可让"读当前值 -> 算增量"也落在锁内(例如计数自增), 避免并发下的 lost update。 """ try: with _usage_lock: data = _read_all() entry = {**_DEFAULT_ENTRY, **data.get(name, {})} resolved = updates(entry) if callable(updates) else updates entry.update(resolved) data[name] = entry _write_all(data) except Exception: logger.debug("Failed to update usage entry for %r", name, exc_info=True) def get_entry(name: str) -> dict: """Return the usage entry for *name*, filling missing fields with defaults.""" data = _read_all() return {**_DEFAULT_ENTRY, **data.get(name, {})} def all_entries() -> dict: """Return all usage entries keyed by skill name.""" return _read_all() def agent_created_entries() -> dict: """Return only entries where source == 'agent' (eligible for curation).""" return {k: v for k, v in _read_all().items() if v.get("source") == SOURCE_AGENT} def bump_view(name: str) -> None: # 用回调:读当前计数 + 自增在同一把锁内完成,避免并发丢计数。 _update_entry(name, lambda e: {"view_count": e.get("view_count", 0) + 1, "last_viewed_at": _now_iso()}) def bump_use(name: str) -> None: _update_entry(name, lambda e: {"use_count": e.get("use_count", 0) + 1, "last_used_at": _now_iso()}) def bump_patch(name: str) -> None: _update_entry(name, lambda e: {"patch_count": e.get("patch_count", 0) + 1, "last_patched_at": _now_iso()}) def set_state(name: str, state: str) -> None: updates: dict = {"state": state} if state == STATE_ARCHIVED: updates["archived_at"] = _now_iso() elif state == STATE_ACTIVE: updates["archived_at"] = None _update_entry(name, updates) def set_pinned(name: str, pinned: bool) -> None: _update_entry(name, {"pinned": pinned}) def set_featured(name: str, featured: bool) -> None: """置顶/取消置顶。与 pinned(防归档)正交,仅影响列表排序展示。""" _update_entry(name, {"featured": featured}) def forget(name: str) -> None: try: with _usage_lock: data = _read_all() data.pop(name, None) _write_all(data) except Exception: logger.debug("Failed to forget usage entry for %r", name, exc_info=True) def mark_agent_created(name: str) -> None: """Record a skill as agent-created; sets source='agent' and initialises created_at.""" _update_entry(name, {"source": SOURCE_AGENT, "created_at": _now_iso(), "state": STATE_ACTIVE}) def mark_uploaded(name: str) -> None: """Record a skill as user-uploaded; sets source='upload'.""" _update_entry(name, {"source": SOURCE_UPLOAD, "created_at": _now_iso()}) def latest_activity_at(name: str) -> str | None: """Return the most recent activity timestamp, falling back to created_at.""" entry = get_entry(name) candidates = [entry.get("last_used_at"), entry.get("last_viewed_at"), entry.get("last_patched_at")] valid = [ts for ts in candidates if ts] if valid: return max(valid) return entry.get("created_at")