fix: use skill activity in curator status

Treat skill views and edits as activity when curator reports and applies lifecycle transitions, so recently loaded or patched skills are not displayed or transitioned as never used.\n\nAdds regression tests for activity derivation, automatic transitions, and CLI status output.
This commit is contained in:
y0shualee 2026-04-30 20:23:17 +08:00 committed by Teknium
parent 564a649e6a
commit f4b76fa272
6 changed files with 193 additions and 16 deletions

View File

@ -7,7 +7,7 @@ daemon): when the agent is idle and the last curator run was longer than
the review.
Responsibilities:
- Auto-transition lifecycle states based on last_used_at timestamps
- Auto-transition lifecycle states based on derived skill activity timestamps
- Spawn a background review agent that can pin / archive / consolidate /
patch agent-created skills via skill_manage
- Persist curator state (last_run_at, paused, etc.) in .curator_state
@ -213,8 +213,8 @@ def should_run_now(now: Optional[datetime] = None) -> bool:
def apply_automatic_transitions(now: Optional[datetime] = None) -> Dict[str, int]:
"""Walk every agent-created skill and move active/stale/archived based on
last_used_at. Pinned skills are never touched. Returns a counter dict
describing what changed."""
the latest real activity timestamp. Pinned skills are never touched.
Returns a counter dict describing what changed."""
from tools import skill_usage as _u
if now is None:
@ -230,10 +230,10 @@ def apply_automatic_transitions(now: Optional[datetime] = None) -> Dict[str, int
if row.get("pinned"):
continue
last_used = _parse_iso(row.get("last_used_at"))
# If never used, treat as using created_at as the anchor so new skills
# don't immediately archive themselves.
anchor = last_used or _parse_iso(row.get("created_at")) or now
last_activity = _parse_iso(row.get("last_activity_at"))
# If never active, treat created_at as the anchor so new skills don't
# immediately archive themselves.
anchor = last_activity or _parse_iso(row.get("created_at")) or now
if anchor.tzinfo is None:
anchor = anchor.replace(tzinfo=timezone.utc)
@ -980,10 +980,11 @@ def _render_candidate_list() -> str:
f"- {r['name']} "
f"state={r['state']} "
f"pinned={'yes' if r.get('pinned') else 'no'} "
f"activity={r.get('activity_count', 0)} "
f"use={r.get('use_count', 0)} "
f"view={r.get('view_count', 0)} "
f"patches={r.get('patch_count', 0)} "
f"last_used={r.get('last_used_at') or 'never'}"
f"last_activity={r.get('last_activity_at') or 'never'}"
)
return "\n".join(lines)

View File

@ -88,16 +88,25 @@ def _cmd_status(args) -> int:
if pinned:
print(f"\npinned ({len(pinned)}): {', '.join(pinned)}")
# Show top 5 least-recently-used active skills
# Show top 5 least-recently-active skills. Views and edits are activity too:
# curator should not report a skill as "never used" right after skill_view()
# or skill_manage() touched it.
active = sorted(
by_state.get("active", []),
key=lambda r: r.get("last_used_at") or r.get("created_at") or "",
key=lambda r: r.get("last_activity_at") or r.get("created_at") or "",
)[:5]
if active:
print("\nleast recently used (top 5):")
print("\nleast recently active (top 5):")
for r in active:
last = _fmt_ts(r.get("last_used_at"))
print(f" {r['name']:40s} use={r.get('use_count', 0):3d} last_used={last}")
last = _fmt_ts(r.get("last_activity_at"))
print(
f" {r['name']:40s} "
f"activity={r.get('activity_count', 0):3d} "
f"use={r.get('use_count', 0):3d} "
f"view={r.get('view_count', 0):3d} "
f"patches={r.get('patch_count', 0):3d} "
f"last_activity={last}"
)
return 0

View File

@ -0,0 +1,56 @@
"""Regression tests for curator skill activity timestamps."""
import importlib
from datetime import datetime, timedelta, timezone
from pathlib import Path
import pytest
def _write_skill(skills_dir: Path, name: str) -> None:
skill_dir = skills_dir / name
skill_dir.mkdir(parents=True, exist_ok=True)
(skill_dir / "SKILL.md").write_text(
f"---\nname: {name}\ndescription: test skill\n---\n\n# {name}\n",
encoding="utf-8",
)
@pytest.fixture
def curator_modules(tmp_path, monkeypatch):
home = tmp_path / ".hermes"
(home / "skills").mkdir(parents=True)
monkeypatch.setenv("HERMES_HOME", str(home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
import tools.skill_usage as skill_usage
import agent.curator as curator
importlib.reload(skill_usage)
importlib.reload(curator)
return home, skill_usage, curator
def test_recent_view_activity_prevents_false_stale_transition(curator_modules, monkeypatch):
home, skill_usage, curator = curator_modules
skills_dir = home / "skills"
_write_skill(skills_dir, "recently-viewed")
now = datetime(2026, 4, 30, tzinfo=timezone.utc)
created_at = (now - timedelta(days=60)).isoformat()
last_viewed_at = (now - timedelta(days=1)).isoformat()
skill_usage.save_usage({
"recently-viewed": {
"created_at": created_at,
"last_viewed_at": last_viewed_at,
"view_count": 1,
"state": "active",
}
})
monkeypatch.setattr(curator, "get_stale_after_days", lambda: 30)
monkeypatch.setattr(curator, "get_archive_after_days", lambda: 90)
counts = curator.apply_automatic_transitions(now=now)
assert counts["marked_stale"] == 0
assert skill_usage.get_record("recently-viewed")["state"] == "active"

View File

@ -0,0 +1,43 @@
"""Tests for the curator CLI status renderer."""
from types import SimpleNamespace
def test_status_uses_last_activity_not_only_last_used(monkeypatch, capsys):
import agent.curator as curator_state
import hermes_cli.curator as curator_cli
import tools.skill_usage as skill_usage
monkeypatch.setattr(curator_state, "load_state", lambda: {
"paused": False,
"last_run_at": None,
"last_run_summary": "(none)",
"run_count": 0,
})
monkeypatch.setattr(curator_state, "is_enabled", lambda: True)
monkeypatch.setattr(curator_state, "get_interval_hours", lambda: 168)
monkeypatch.setattr(curator_state, "get_stale_after_days", lambda: 30)
monkeypatch.setattr(curator_state, "get_archive_after_days", lambda: 90)
monkeypatch.setattr(skill_usage, "agent_created_report", lambda: [
{
"name": "recently-viewed",
"state": "active",
"pinned": False,
"use_count": 0,
"view_count": 3,
"patch_count": 1,
"created_at": "2026-01-01T00:00:00+00:00",
"last_used_at": None,
"last_viewed_at": "2026-04-30T10:00:00+00:00",
"last_patched_at": "2026-04-30T11:00:00+00:00",
"last_activity_at": "2026-04-30T11:00:00+00:00",
"activity_count": 4,
}
])
assert curator_cli._cmd_status(SimpleNamespace()) == 0
out = capsys.readouterr().out
assert "least recently active" in out
assert "activity= 4" in out
assert "last_activity=never" not in out
assert "last_used=never" not in out

View File

@ -401,6 +401,26 @@ def test_agent_created_report_excludes_bundled_and_hub(skills_home):
assert "hubbed" not in names
def test_agent_created_report_derives_activity_from_view_and_patch(skills_home, monkeypatch):
import tools.skill_usage as skill_usage
skills_dir = skills_home / "skills"
_write_skill(skills_dir, "mine")
timestamps = iter([
"2026-04-30T10:00:00+00:00",
"2026-04-30T11:00:00+00:00",
"2026-04-30T12:00:00+00:00",
"2026-04-30T13:00:00+00:00",
])
monkeypatch.setattr(skill_usage, "_now_iso", lambda: next(timestamps))
skill_usage.bump_view("mine")
skill_usage.bump_patch("mine")
row = next(r for r in skill_usage.agent_created_report() if r["name"] == "mine")
assert row["activity_count"] == 2
assert row["last_activity_at"] == "2026-04-30T12:00:00+00:00"
# ---------------------------------------------------------------------------
# Provenance guard — telemetry must not leak records for bundled/hub skills

View File

@ -2,7 +2,8 @@
Tracks per-skill usage metadata in a sidecar JSON file (~/.hermes/skills/.usage.json)
keyed by skill name. Counters are bumped by the existing skill tools (skill_view,
skill_manage); the curator orchestrator reads them to decide lifecycle transitions.
skill_manage); the curator orchestrator reads the derived activity timestamp to
decide lifecycle transitions.
Design notes:
- Sidecar, not frontmatter. Keeps operational telemetry out of user-authored
@ -57,6 +58,50 @@ def _now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
def _parse_iso_timestamp(value: Any) -> Optional[datetime]:
"""Parse an ISO timestamp defensively for activity comparisons."""
if not value:
return None
try:
parsed = datetime.fromisoformat(str(value))
except (TypeError, ValueError):
return None
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
return parsed
def latest_activity_at(record: Dict[str, Any]) -> Optional[str]:
"""Return the newest actual activity timestamp for a usage record.
"Activity" means a skill was used, viewed, or patched. Creation time is
intentionally excluded so callers can still distinguish never-active skills;
lifecycle code can fall back to ``created_at`` as its own anchor.
"""
latest_dt: Optional[datetime] = None
latest_raw: Optional[str] = None
for key in ("last_used_at", "last_viewed_at", "last_patched_at"):
raw = record.get(key)
dt = _parse_iso_timestamp(raw)
if dt is None:
continue
if latest_dt is None or dt > latest_dt:
latest_dt = dt
latest_raw = str(raw)
return latest_raw
def activity_count(record: Dict[str, Any]) -> int:
"""Return the total observed activity count across use/view/patch events."""
total = 0
for key in ("use_count", "view_count", "patch_count"):
try:
total += int(record.get(key) or 0)
except (TypeError, ValueError):
continue
return total
# ---------------------------------------------------------------------------
# Provenance — which skills are agent-created (and thus eligible for curation)
# ---------------------------------------------------------------------------
@ -442,7 +487,7 @@ 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_used_at, use_count, ...}
"""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."""
data = load_usage()
@ -454,5 +499,8 @@ def agent_created_report() -> List[Dict[str, Any]]:
base = _empty_record()
for k, v in base.items():
rec.setdefault(k, v)
rows.append({"name": name, **rec})
row = {"name": name, **rec}
row["last_activity_at"] = latest_activity_at(row)
row["activity_count"] = activity_count(row)
rows.append(row)
return rows