289 lines
10 KiB
Python
289 lines
10 KiB
Python
"""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 collections import OrderedDict
|
|
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()
|
|
|
|
# One user run can first load SKILL.md and then execute several helper scripts.
|
|
# Those are implementation details of one logical skill invocation, so retain a
|
|
# small process-local LRU keyed by (run, skill) and increment the public counter
|
|
# only once. A run stays on one worker; persistence across restarts is neither
|
|
# needed nor desirable here.
|
|
_MAX_RECENT_USE_EVENTS = 8192
|
|
_recent_use_events: OrderedDict[tuple[str, str], None] = OrderedDict()
|
|
|
|
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()})
|
|
|
|
|
|
async def _increment_use_projection(name: str, use_count: int, used_at: datetime) -> None:
|
|
"""Keep the MySQL skill-list projection current without importing ``app``."""
|
|
try:
|
|
from sqlalchemy import case, update
|
|
|
|
from deerflow.persistence.engine import get_session_factory
|
|
from deerflow.persistence.skills.model import SkillRow
|
|
|
|
session_factory = get_session_factory()
|
|
if session_factory is None:
|
|
return
|
|
async with session_factory() as session:
|
|
await session.execute(
|
|
update(SkillRow)
|
|
.where(SkillRow.name == name)
|
|
.values(
|
|
# Fire-and-forget writes may finish out of order. Never let
|
|
# an older absolute sidecar count overwrite a newer value.
|
|
use_count=case((SkillRow.use_count < use_count, use_count), else_=SkillRow.use_count),
|
|
last_activity_at=case(
|
|
(SkillRow.last_activity_at.is_(None), used_at),
|
|
(SkillRow.last_activity_at < used_at, used_at),
|
|
else_=SkillRow.last_activity_at,
|
|
),
|
|
updated_at=case(
|
|
(SkillRow.updated_at < used_at, used_at),
|
|
else_=SkillRow.updated_at,
|
|
),
|
|
)
|
|
)
|
|
await session.commit()
|
|
except Exception:
|
|
logger.debug("Failed to update skill usage projection for %r", name, exc_info=True)
|
|
|
|
|
|
def _schedule_use_projection_increment(name: str, use_count: int, used_at: datetime) -> None:
|
|
"""Best-effort bridge from synchronous sandbox tools to the DB home loop."""
|
|
try:
|
|
from deerflow.persistence.engine import fire_and_forget_db_write
|
|
|
|
fire_and_forget_db_write(_increment_use_projection(name, use_count, used_at))
|
|
except Exception:
|
|
logger.debug("Failed to schedule skill usage projection for %r", name, exc_info=True)
|
|
|
|
|
|
def runtime_use_event_key(runtime: Any | None) -> str | None:
|
|
"""Return the stable user-run identifier carried by ``ToolRuntime``."""
|
|
if runtime is None:
|
|
return None
|
|
context = getattr(runtime, "context", None)
|
|
if isinstance(context, dict):
|
|
for key in ("run_id", "langgraph_run_id"):
|
|
value = context.get(key)
|
|
if value is not None and str(value).strip():
|
|
return str(value).strip()
|
|
config = getattr(runtime, "config", None)
|
|
if isinstance(config, dict):
|
|
for bucket_name in ("configurable", "metadata"):
|
|
bucket = config.get(bucket_name)
|
|
if not isinstance(bucket, dict):
|
|
continue
|
|
for key in ("run_id", "langgraph_run_id"):
|
|
value = bucket.get(key)
|
|
if value is not None and str(value).strip():
|
|
return str(value).strip()
|
|
return None
|
|
|
|
|
|
def bump_use(name: str, *, event_key: str | None = None) -> bool:
|
|
"""Count one logical use, optionally de-duplicated by user run.
|
|
|
|
Returns ``True`` only when the counter changed. This makes callers and tests
|
|
able to distinguish a new invocation from a second file/script access in
|
|
the same run.
|
|
"""
|
|
normalized_name = str(name or "").strip()
|
|
if not normalized_name:
|
|
return False
|
|
normalized_event = str(event_key or "").strip() or None
|
|
dedupe_key = (normalized_event, normalized_name) if normalized_event else None
|
|
used_at = datetime.now(UTC)
|
|
new_count = 0
|
|
try:
|
|
with _usage_lock:
|
|
if dedupe_key is not None and dedupe_key in _recent_use_events:
|
|
_recent_use_events.move_to_end(dedupe_key)
|
|
return False
|
|
data = _read_all()
|
|
entry = {**_DEFAULT_ENTRY, **data.get(normalized_name, {})}
|
|
new_count = max(0, int(entry.get("use_count") or 0)) + 1
|
|
entry.update({"use_count": new_count, "last_used_at": used_at.isoformat()})
|
|
data[normalized_name] = entry
|
|
_write_all(data)
|
|
if dedupe_key is not None:
|
|
_recent_use_events[dedupe_key] = None
|
|
_recent_use_events.move_to_end(dedupe_key)
|
|
while len(_recent_use_events) > _MAX_RECENT_USE_EVENTS:
|
|
_recent_use_events.popitem(last=False)
|
|
except Exception:
|
|
logger.debug("Failed to bump usage entry for %r", normalized_name, exc_info=True)
|
|
return False
|
|
|
|
_schedule_use_projection_increment(normalized_name, new_count, used_at)
|
|
return True
|
|
|
|
|
|
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")
|