seb/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(curator): prune built-in skills after inactivity + track usage for all skills (#36701)

Browse Source 
Two related changes to the skill curator:

1. Built-in pruning. New curator.prune_builtins config (default on) lets the
   curator archive bundled built-in skills after the inactivity period, not
   just agent-created ones. A .curator_suppressed list tells the update-time
   re-seeder (tools/skills_sync) to leave pruned built-ins archived, so the
   prune is durable across `hermes update`. Built-ins are seeded with a
   baseline record on first sight, so the inactivity clock starts at upgrade
   time -- no mass-prune on the first run. Hub-installed skills are never
   pruned regardless of the flag. Restoring a built-in clears its suppression.

2. Usage tracking for all skills. Telemetry (view/use/patch) was wrongly gated
   behind curation-eligibility, so built-ins were tracked only when prunable
   and hub skills never. Telemetry is observability and is now decoupled from
   curation: every skill accrues usage counts regardless of provenance, while
   lifecycle mutators (set_state/set_pinned/mark_agent_created) stay
   curation-gated. New usage_report() + provenance() expose all skills with an
   agent/bundled/hub tag.
This commit is contained in:
Teknium
2026-06-01 02:07:32 -07:00
committed by GitHub
parent 0622a70eb4
commit 70e1571d89
8 changed files with 622 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/skill_usage.py
+285 -41
View File
@@ -217,21 +217,111 @@ def _read_hub_installed_names() -> Set[str]:
return set()
def list_agent_created_skill_names() -> List[str]:
"""Enumerate skills explicitly authored by the agent.
def _prune_builtins_enabled() -> bool:
"""Whether bundled built-in skills are eligible for curator pruning.
The curator operates exclusively on this set. Skills are only eligible
after ``skill_manage(action="create")`` marks them in ``.usage.json``;
manually authored skills must not be inferred from filesystem location.
Bundled / hub skills are maintained by their upstream sources and must
never be pruned here.
Reads ``curator.prune_builtins`` from config (default True). Lazy import
keeps this module importable without the CLI config layer (e.g. in the
update/sync context); on any failure we fall back to the default. The real
safety against a mass-prune is the curator's seed-on-first-sight, not this
flag — built-ins only archive after a fresh inactivity window.
"""
try:
from hermes_cli.config import load_config
cfg = load_config()
cur = cfg.get("curator") if isinstance(cfg, dict) else None
if isinstance(cur, dict):
return bool(cur.get("prune_builtins", True))
except Exception as e: # pragma: no cover — best-effort config read
logger.debug("Failed to read curator.prune_builtins: %s", e)
return True
def _suppressed_file() -> Path:
return _skills_dir() / ".curator_suppressed"
def read_suppressed_names() -> Set[str]:
"""Built-in skills the curator pruned — the re-seeder must leave archived.
One skill name per line in ``~/.hermes/skills/.curator_suppressed``. This is
what makes pruning a built-in durable: without it, ``hermes update`` would
re-copy the bundled skill on the next sync.
"""
path = _suppressed_file()
if not path.exists():
return set()
names: Set[str] = set()
try:
for line in path.read_text(encoding="utf-8").splitlines():
line = line.strip()
if line and not line.startswith("#"):
names.add(line)
except OSError as e:
logger.debug("Failed to read curator suppression list: %s", e)
return names
def _write_suppressed_names(names: Set[str]) -> None:
path = _suppressed_file()
try:
path.parent.mkdir(parents=True, exist_ok=True)
data = "\n".join(sorted(names)) + ("\n" if names else "")
fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=".curator_suppressed_", suffix=".tmp")
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
f.write(data)
f.flush()
os.fsync(f.fileno())
os.replace(tmp, path)
except BaseException:
try:
os.unlink(tmp)
except OSError:
pass
raise
except Exception as e:
logger.debug("Failed to write curator suppression list: %s", e, exc_info=True)
def add_suppressed_name(skill_name: str) -> None:
"""Record that a built-in skill was pruned, so sync won't restore it."""
if not skill_name:
return
names = read_suppressed_names()
if skill_name not in names:
names.add(skill_name)
_write_suppressed_names(names)
def remove_suppressed_name(skill_name: str) -> None:
"""Clear a built-in's suppression entry (e.g. on restore)."""
if not skill_name:
return
names = read_suppressed_names()
if skill_name in names:
names.discard(skill_name)
_write_suppressed_names(names)
def list_agent_created_skill_names() -> List[str]:
"""Enumerate skills the curator may manage.
Always includes agent-authored skills (those marked in ``.usage.json`` via
``skill_manage(action="create")``). When ``curator.prune_builtins`` is
enabled, bundled built-in skills are ALSO included even though they have no
agent-created usage record — their inactivity clock is anchored on first
sight (see ``apply_automatic_transitions``). Hub-installed skills are never
included; manually authored skills are not inferred from filesystem
location.
"""
base = _skills_dir()
if not base.exists():
return []
bundled = _read_bundled_manifest_names()
hub = _read_hub_installed_names()
off_limits = bundled | hub
bundled = _read_bundled_manifest_names()
prune_builtins = _prune_builtins_enabled()
usage = load_usage()
names: List[str] = []
@@ -241,12 +331,21 @@ def list_agent_created_skill_names() -> List[str]:
if is_excluded_skill_path(skill_md):
continue
try:
rel = skill_md.relative_to(base)
skill_md.relative_to(base)
except ValueError:
continue
name = _read_skill_name(skill_md, fallback=skill_md.parent.name)
if name in off_limits:
# Hub-installed skills are always off-limits.
if name in hub:
continue
if name in bundled:
# Built-ins are only candidates when pruning is enabled. They never
# carry a curator-managed record, so the record gate is skipped.
if not prune_builtins:
continue
names.append(name)
continue
# Agent-authored (or local-manual) skills must opt in via their record.
if not _is_curator_managed_record(usage.get(name)):
continue
names.append(name)
@@ -293,6 +392,30 @@ def is_agent_created(skill_name: str) -> bool:
return skill_name not in off_limits
def is_hub_installed(skill_name: str) -> bool:
"""Whether *skill_name* was installed via the Skills Hub."""
return skill_name in _read_hub_installed_names()
def is_bundled(skill_name: str) -> bool:
"""Whether *skill_name* was seeded from the bundled repo skills."""
return skill_name in _read_bundled_manifest_names()
def is_curation_eligible(skill_name: str) -> bool:
"""Whether the curator may track/archive *skill_name*.
Agent-created skills are always eligible. Bundled built-ins become eligible
only when ``curator.prune_builtins`` is enabled. Hub-installed skills are
NEVER eligible — they have an external upstream owner.
"""
if is_hub_installed(skill_name):
return False
if is_bundled(skill_name):
return _prune_builtins_enabled()
return True
def _is_curator_managed_record(record: Any) -> bool:
"""Return True when a usage record opts a skill into curator management."""
if not isinstance(record, dict):
@@ -377,17 +500,43 @@ def get_record(skill_name: str) -> Dict[str, Any]:
return rec
def _mutate(skill_name: str, mutator) -> None:
def seed_record_if_missing(skill_name: str) -> None:
"""Persist a baseline usage record for a curation-eligible skill.
Built-ins carry no usage record until something touches them, which leaves
their inactivity clock with no anchor. Seeding a record here fixes
``created_at`` to the moment the curator first sees the skill, so the
archive/stale clock measures non-use FROM THEN — not from epoch. No-op when
a record already exists or the skill isn't curation-eligible.
"""
if not skill_name or not is_curation_eligible(skill_name):
return
try:
with _usage_file_lock():
data = load_usage()
if isinstance(data.get(skill_name), dict):
return
data[skill_name] = _empty_record()
save_usage(data)
except Exception as e:
logger.debug("skill_usage.seed_record_if_missing(%s) failed: %s", skill_name, e, exc_info=True)
def _mutate(skill_name: str, mutator, *, require_curation_eligible: bool = False) -> None:
"""Load, apply *mutator(record)* in place, save. Best-effort.
Bundled and hub-installed skills are NEVER recorded in the sidecar.
Local manual skills may still accrue usage telemetry, but they only
become curator-managed when ``created_by`` is explicitly marked.
By default this records telemetry for ANY skill — bundled, hub-installed,
or agent-created — because usage tracking is pure observability and is
orthogonal to whether a skill is ever curated. Lifecycle mutators
(``set_state``, ``set_pinned``, ``mark_agent_created``) pass
``require_curation_eligible=True`` so they never write meaningless state
onto a skill the curator can't manage (e.g. an ``archived`` flag on a
hub-installed skill).
"""
if not skill_name:
return
try:
if not is_agent_created(skill_name):
if require_curation_eligible and not is_curation_eligible(skill_name):
return
with _usage_file_lock():
data = load_usage()
@@ -402,11 +551,15 @@ def _mutate(skill_name: str, mutator) -> None:
# ---------------------------------------------------------------------------
# Public counter-bump helpers
# Public counter-bump helpers — telemetry for ALL skills (observability only)
# ---------------------------------------------------------------------------
def bump_view(skill_name: str) -> None:
"""Bump view_count and last_viewed_at. Called from skill_view()."""
"""Bump view_count and last_viewed_at. Called from skill_view().
Tracks every skill regardless of provenance — built-ins and hub skills
included. Usage telemetry is observability, not a curation signal.
"""
def _apply(rec: Dict[str, Any]) -> None:
rec["view_count"] = int(rec.get("view_count") or 0) + 1
rec["last_viewed_at"] = _now_iso()
@@ -415,7 +568,10 @@ def bump_view(skill_name: str) -> None:
def bump_use(skill_name: str) -> None:
"""Bump use_count and last_used_at. Called when a skill is actively used
(e.g. loaded into the prompt path or referenced from an assistant turn)."""
(e.g. loaded into the prompt path or referenced from an assistant turn).
Tracks every skill regardless of provenance.
"""
def _apply(rec: Dict[str, Any]) -> None:
rec["use_count"] = int(rec.get("use_count") or 0) + 1
rec["last_used_at"] = _now_iso()
@@ -423,7 +579,10 @@ def bump_use(skill_name: str) -> None:
def bump_patch(skill_name: str) -> None:
"""Bump patch_count and last_patched_at. Called from skill_manage (patch/edit)."""
"""Bump patch_count and last_patched_at. Called from skill_manage (patch/edit).
Tracks every skill regardless of provenance.
"""
def _apply(rec: Dict[str, Any]) -> None:
rec["patch_count"] = int(rec.get("patch_count") or 0) + 1
rec["last_patched_at"] = _now_iso()
@@ -438,11 +597,12 @@ def mark_agent_created(skill_name: str) -> None:
"""
def _apply(rec: Dict[str, Any]) -> None:
rec["created_by"] = "agent"
_mutate(skill_name, _apply)
_mutate(skill_name, _apply, require_curation_eligible=True)
def set_state(skill_name: str, state: str) -> None:
"""Set lifecycle state. No-op if *state* is invalid."""
"""Set lifecycle state. No-op if *state* is invalid or the skill isn't
curator-manageable (hub skills, or built-ins with pruning disabled)."""
if state not in _VALID_STATES:
logger.debug("set_state: invalid state %r for %s", state, skill_name)
return
@@ -452,13 +612,13 @@ def set_state(skill_name: str, state: str) -> None:
rec["archived_at"] = _now_iso()
elif state == STATE_ACTIVE:
rec["archived_at"] = None
_mutate(skill_name, _apply)
_mutate(skill_name, _apply, require_curation_eligible=True)
def set_pinned(skill_name: str, pinned: bool) -> None:
def _apply(rec: Dict[str, Any]) -> None:
rec["pinned"] = bool(pinned)
_mutate(skill_name, _apply)
_mutate(skill_name, _apply, require_curation_eligible=True)
def forget(skill_name: str) -> None:
@@ -480,13 +640,20 @@ def forget(skill_name: str) -> None:
# ---------------------------------------------------------------------------
def archive_skill(skill_name: str) -> Tuple[bool, str]:
"""Move an agent-created skill directory to ~/.hermes/skills/.archive/.
"""Move a curator-eligible skill directory to ~/.hermes/skills/.archive/.
Returns (ok, message). Never archives bundled or hub skills — callers are
responsible for checking provenance, but we double-check here as a safety net.
Returns (ok, message). Never archives hub-installed skills. Bundled
built-ins are only archivable when ``curator.prune_builtins`` is enabled;
when one is archived, its name is added to the suppression list so the
update-time re-seeder leaves it archived instead of restoring it.
"""
if not is_agent_created(skill_name):
return False, f"skill '{skill_name}' is bundled or hub-installed; never archive"
if not is_curation_eligible(skill_name):
if is_hub_installed(skill_name):
return False, f"skill '{skill_name}' is hub-installed; never archive"
return False, (
f"skill '{skill_name}' is a bundled built-in; enable "
"curator.prune_builtins to allow pruning it"
)
skill_dir = _find_skill_dir(skill_name)
if skill_dir is None:
@@ -514,6 +681,10 @@ def archive_skill(skill_name: str) -> Tuple[bool, str]:
except Exception as e2:
return False, f"failed to archive: {e2}"
# Pruning a built-in only sticks if the re-seeder is told to leave it alone.
if is_bundled(skill_name):
add_suppressed_name(skill_name)
set_state(skill_name, STATE_ARCHIVED)
return True, f"archived to {dest}"
@@ -522,14 +693,24 @@ def restore_skill(skill_name: str) -> Tuple[bool, str]:
"""Move an archived skill back to ~/.hermes/skills/. Restores to the flat
top-level layout; original category nesting is NOT reconstructed.
Refuses to restore under a name that now collides with a bundled or
hub-installed skill — that would shadow the upstream version.
Refuses to restore under a name that now collides with a hub-installed
skill — that would shadow the upstream version. Also refuses to restore
over a bundled built-in UNLESS ``curator.prune_builtins`` is enabled (in
which case built-ins are curator-managed and restoring is the documented
way to lift a prune). Restoring clears any suppression entry so future
updates may re-seed the built-in again.
"""
# If a bundled or hub skill has since been installed under the same
# name, refuse to restore rather than shadow it.
if not is_agent_created(skill_name):
# Hub skills always have an external upstream owner — never shadow them.
if is_hub_installed(skill_name):
return False, (
f"skill '{skill_name}' is now bundled or hub-installed; "
f"skill '{skill_name}' is now hub-installed; "
"restore would shadow the upstream version"
)
# A bundled built-in is upstream-owned UNLESS prune_builtins is on. With the
# flag off, restoring over it would shadow the bundled version.
if is_bundled(skill_name) and not _prune_builtins_enabled():
return False, (
f"skill '{skill_name}' is now bundled; "
"restore would shadow the upstream version"
)
archive_root = _archive_dir()
@@ -563,6 +744,9 @@ def restore_skill(skill_name: str) -> Tuple[bool, str]:
except Exception as e:
return False, f"failed to restore: {e}"
# Restoring a pruned built-in lifts its suppression so updates can manage it.
remove_suppressed_name(skill_name)
set_state(skill_name, STATE_ACTIVE)
return True, f"restored to {dest}"
@@ -590,19 +774,79 @@ def _find_skill_dir(skill_name: str) -> Optional[Path]:
def agent_created_report() -> List[Dict[str, Any]]:
"""Return a list of {name, state, pinned, last_activity_at, ...}
records for every agent-created skill. Missing usage records are backfilled
with defaults so callers can always index fields."""
records for every curator-managed skill. Missing usage records are
backfilled with defaults so callers can always index fields.
Each row carries ``_persisted``: True when a real record exists in
``.usage.json``, False when the row is a fresh backfill (e.g. a built-in
seen for the first time). The curator uses this to seed the inactivity
clock instead of treating an unrecorded skill as ancient.
"""
data = load_usage()
rows: List[Dict[str, Any]] = []
for name in list_agent_created_skill_names():
rec = data.get(name)
if not isinstance(rec, dict):
rec = _empty_record()
raw = data.get(name)
persisted = isinstance(raw, dict)
rec: Dict[str, Any] = raw if isinstance(raw, dict) else _empty_record()
base = _empty_record()
for k, v in base.items():
rec.setdefault(k, v)
row = {"name": name, **rec}
row = {"name": name, **rec, "_persisted": persisted}
row["last_activity_at"] = latest_activity_at(row)
row["activity_count"] = activity_count(row)
rows.append(row)
return rows
def provenance(skill_name: str) -> str:
"""Classify a skill's origin: 'hub', 'bundled', or 'agent'.
'agent' covers both agent-authored and local manually-authored skills —
anything not seeded from the bundled repo or installed via the hub.
"""
if is_hub_installed(skill_name):
return "hub"
if is_bundled(skill_name):
return "bundled"
return "agent"
def usage_report() -> List[Dict[str, Any]]:
"""Return usage telemetry for EVERY skill on disk, with provenance.
Unlike ``agent_created_report()`` (which is scoped to curator-managed
candidates), this surfaces all skills — bundled built-ins and
hub-installed included — so callers can answer "how often is this skill
used" independent of whether it's ever curated. Rows carry a
``provenance`` field ('agent' | 'bundled' | 'hub') and ``_persisted``
(whether a real ``.usage.json`` record backs the row).
"""
base = _skills_dir()
if not base.exists():
return []
data = load_usage()
rows: List[Dict[str, Any]] = []
seen: set = set()
for skill_md in base.rglob("SKILL.md"):
if is_excluded_skill_path(skill_md):
continue
name = _read_skill_name(skill_md, fallback=skill_md.parent.name)
if name in seen:
continue
seen.add(name)
raw = data.get(name)
persisted = isinstance(raw, dict)
rec: Dict[str, Any] = raw if isinstance(raw, dict) else _empty_record()
base_rec = _empty_record()
for k, v in base_rec.items():
rec.setdefault(k, v)
row = {
"name": name,
**rec,
"provenance": provenance(name),
"_persisted": persisted,
}
row["last_activity_at"] = latest_activity_at(row)
row["activity_count"] = activity_count(row)
rows.append(row)
return sorted(rows, key=lambda r: r["name"])