deerflow-code/offline-backend-20260512/backend/packages/harness/deerflow/skills/usage.py

"""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")