Anton Palgunov
f936968c5c
- agent/session_stats.py: SessionDB context/compression metrics - agent/skill_stats.py: curator usage.json reader + prune history - agent/system_health.py: gateway uptime, version, cron activity - agent/stats_dashboard.py: Telegram-friendly bullet renderer - cli.py: /stats dispatch + _handle_stats_command method - gateway/run.py: /stats dispatch + _handle_stats_command for messaging platforms - hermes_cli/commands.py: /stats CommandDef registration
85 lines
2.9 KiB
Python
85 lines
2.9 KiB
Python
"""Skill and curator telemetry collectors for the /stats dashboard."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List
|
|
|
|
from hermes_constants import get_hermes_home
|
|
from tools.skill_usage import (
|
|
STATE_ARCHIVED,
|
|
activity_count,
|
|
latest_activity_at,
|
|
load_usage,
|
|
)
|
|
|
|
|
|
def _parse_dt(value: Any):
|
|
if not value:
|
|
return None
|
|
try:
|
|
dt = datetime.fromisoformat(str(value))
|
|
except (TypeError, ValueError):
|
|
return None
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
return dt
|
|
|
|
|
|
def _activity_count(record: Dict[str, Any]) -> int:
|
|
return activity_count(record)
|
|
|
|
|
|
def collect_skill_stats(limit: int = 5) -> Dict[str, Any]:
|
|
usage = load_usage()
|
|
rows: List[Dict[str, Any]] = []
|
|
for name, record in usage.items():
|
|
if not isinstance(record, dict):
|
|
continue
|
|
count = _activity_count(record)
|
|
rows.append({
|
|
"name": str(name),
|
|
"activity_count": count,
|
|
"use_count": int(record.get("use_count") or 0),
|
|
"view_count": int(record.get("view_count") or 0),
|
|
"patch_count": int(record.get("patch_count") or 0),
|
|
"last_activity_at": latest_activity_at(record),
|
|
"state": record.get("state") or "active",
|
|
})
|
|
rows.sort(key=lambda r: (r["activity_count"], r["name"]), reverse=True)
|
|
return {"top_skills": rows[:limit], "usage_records": len(rows)}
|
|
|
|
|
|
def collect_curator_prunes(days: int = 7, limit: int = 3) -> Dict[str, Any]:
|
|
cutoff = datetime.now(timezone.utc) - timedelta(days=days)
|
|
usage = load_usage()
|
|
archived = []
|
|
for name, record in usage.items():
|
|
if not isinstance(record, dict):
|
|
continue
|
|
if record.get("state") != STATE_ARCHIVED:
|
|
continue
|
|
ts = record.get("archived_at") or record.get("last_patched_at") or record.get("created_at")
|
|
dt = _parse_dt(ts)
|
|
if dt is not None and dt < cutoff:
|
|
continue
|
|
archived.append({"name": str(name), "archived_at": ts})
|
|
|
|
# Also inspect the archive directory so manually restored/old usage sidecars
|
|
# still have a real filesystem source for the dashboard.
|
|
archive_dir = get_hermes_home() / "skills" / ".archive"
|
|
if archive_dir.exists():
|
|
for path in archive_dir.iterdir():
|
|
if not path.is_dir():
|
|
continue
|
|
try:
|
|
dt = datetime.fromtimestamp(path.stat().st_mtime, timezone.utc)
|
|
except OSError:
|
|
continue
|
|
if dt >= cutoff and not any(row["name"] == path.name for row in archived):
|
|
archived.append({"name": path.name, "archived_at": dt.isoformat()})
|
|
|
|
archived.sort(key=lambda r: str(r.get("archived_at") or ""), reverse=True)
|
|
return {"recent_prunes": archived[:limit], "recent_prune_count": len(archived), "days": days}
|