test_skill_usage.py
1 """Tests for tools/skill_usage.py — sidecar telemetry + provenance filtering.""" 2 3 import json 4 import os 5 from pathlib import Path 6 7 import pytest 8 9 10 @pytest.fixture 11 def skills_home(tmp_path, monkeypatch): 12 """Isolated HERMES_HOME with a clean skills/ dir for each test.""" 13 home = tmp_path / ".hermes" 14 home.mkdir() 15 (home / "skills").mkdir() 16 monkeypatch.setattr(Path, "home", lambda: tmp_path) 17 monkeypatch.setenv("HERMES_HOME", str(home)) 18 # Force skill_usage module to re-resolve paths per test 19 import importlib 20 import tools.skill_usage as mod 21 importlib.reload(mod) 22 return home 23 24 25 def _write_skill(skills_dir: Path, name: str, category: str = ""): 26 """Create a minimal SKILL.md with a name: frontmatter field.""" 27 if category: 28 d = skills_dir / category / name 29 else: 30 d = skills_dir / name 31 d.mkdir(parents=True, exist_ok=True) 32 (d / "SKILL.md").write_text( 33 f"""--- 34 name: {name} 35 description: test skill 36 --- 37 38 # body 39 """, 40 encoding="utf-8", 41 ) 42 return d 43 44 45 # --------------------------------------------------------------------------- 46 # Round-trip 47 # --------------------------------------------------------------------------- 48 49 def test_empty_usage_returns_empty_dict(skills_home): 50 from tools.skill_usage import load_usage 51 assert load_usage() == {} 52 53 54 def test_save_and_load_roundtrip(skills_home): 55 from tools.skill_usage import load_usage, save_usage 56 data = {"skill-a": {"use_count": 3, "state": "active"}} 57 save_usage(data) 58 loaded = load_usage() 59 assert loaded["skill-a"]["use_count"] == 3 60 assert loaded["skill-a"]["state"] == "active" 61 62 63 def test_save_is_atomic_no_partial_tmp_files(skills_home): 64 from tools.skill_usage import save_usage, _usage_file 65 save_usage({"x": {"use_count": 1}}) 66 skills_dir = _usage_file().parent 67 # No leftover tempfile 68 for p in skills_dir.iterdir(): 69 assert not p.name.startswith(".usage_"), f"leftover tmp: {p.name}" 70 71 72 def test_get_record_missing_returns_empty_record(skills_home): 73 from tools.skill_usage import get_record 74 rec = get_record("nonexistent") 75 assert rec["use_count"] == 0 76 assert rec["view_count"] == 0 77 assert rec["state"] == "active" 78 assert rec["pinned"] is False 79 assert rec["archived_at"] is None 80 81 82 def test_get_record_backfills_missing_keys(skills_home): 83 from tools.skill_usage import get_record, save_usage 84 save_usage({"legacy": {"use_count": 5}}) # old-format record 85 rec = get_record("legacy") 86 assert rec["use_count"] == 5 87 assert "view_count" in rec # backfilled 88 assert "state" in rec 89 90 91 def test_load_usage_handles_corrupt_file(skills_home): 92 from tools.skill_usage import load_usage, _usage_file 93 _usage_file().write_text("{ not json }", encoding="utf-8") 94 assert load_usage() == {} 95 96 97 # --------------------------------------------------------------------------- 98 # Counter bumps 99 # --------------------------------------------------------------------------- 100 101 def test_bump_view_increments_and_timestamps(skills_home): 102 from tools.skill_usage import bump_view, get_record 103 bump_view("my-skill") 104 bump_view("my-skill") 105 rec = get_record("my-skill") 106 assert rec["view_count"] == 2 107 assert rec["last_viewed_at"] is not None 108 109 110 def test_bump_use_increments_and_timestamps(skills_home): 111 from tools.skill_usage import bump_use, get_record 112 bump_use("my-skill") 113 rec = get_record("my-skill") 114 assert rec["use_count"] == 1 115 assert rec["last_used_at"] is not None 116 117 118 def test_bump_patch_increments_and_timestamps(skills_home): 119 from tools.skill_usage import bump_patch, get_record 120 bump_patch("my-skill") 121 rec = get_record("my-skill") 122 assert rec["patch_count"] == 1 123 assert rec["last_patched_at"] is not None 124 125 126 def test_bump_on_empty_name_is_noop(skills_home): 127 from tools.skill_usage import bump_view, load_usage 128 bump_view("") 129 assert load_usage() == {} 130 131 132 def test_bumps_do_not_corrupt_other_skills(skills_home): 133 from tools.skill_usage import bump_view, bump_use, get_record 134 bump_view("skill-a") 135 bump_use("skill-b") 136 bump_view("skill-a") 137 assert get_record("skill-a")["view_count"] == 2 138 assert get_record("skill-a")["use_count"] == 0 139 assert get_record("skill-b")["use_count"] == 1 140 141 142 # --------------------------------------------------------------------------- 143 # State transitions 144 # --------------------------------------------------------------------------- 145 146 def test_set_state_active(skills_home): 147 from tools.skill_usage import set_state, get_record, STATE_ACTIVE 148 set_state("x", STATE_ACTIVE) 149 assert get_record("x")["state"] == "active" 150 151 152 def test_set_state_archived_records_timestamp(skills_home): 153 from tools.skill_usage import set_state, get_record, STATE_ARCHIVED 154 set_state("x", STATE_ARCHIVED) 155 rec = get_record("x") 156 assert rec["state"] == "archived" 157 assert rec["archived_at"] is not None 158 159 160 def test_set_state_invalid_is_noop(skills_home): 161 from tools.skill_usage import set_state, get_record 162 set_state("x", "bogus") 163 # No record created for invalid state 164 rec = get_record("x") 165 assert rec["state"] == "active" # default 166 167 168 def test_restoring_from_archive_clears_timestamp(skills_home): 169 from tools.skill_usage import set_state, get_record, STATE_ARCHIVED, STATE_ACTIVE 170 set_state("x", STATE_ARCHIVED) 171 assert get_record("x")["archived_at"] is not None 172 set_state("x", STATE_ACTIVE) 173 assert get_record("x")["archived_at"] is None 174 175 176 def test_set_pinned(skills_home): 177 from tools.skill_usage import set_pinned, get_record 178 set_pinned("x", True) 179 assert get_record("x")["pinned"] is True 180 set_pinned("x", False) 181 assert get_record("x")["pinned"] is False 182 183 184 def test_forget_removes_record(skills_home): 185 from tools.skill_usage import bump_view, forget, load_usage 186 bump_view("x") 187 assert "x" in load_usage() 188 forget("x") 189 assert "x" not in load_usage() 190 191 192 # --------------------------------------------------------------------------- 193 # Provenance filter — the load-bearing safety check 194 # --------------------------------------------------------------------------- 195 196 def test_agent_created_excludes_bundled(skills_home): 197 from tools.skill_usage import list_agent_created_skill_names, mark_agent_created 198 skills_dir = skills_home / "skills" 199 _write_skill(skills_dir, "bundled-skill", category="github") 200 _write_skill(skills_dir, "my-skill") 201 mark_agent_created("my-skill") 202 # Seed a bundled manifest marking bundled-skill as upstream 203 (skills_dir / ".bundled_manifest").write_text( 204 "bundled-skill:abc123\n", encoding="utf-8", 205 ) 206 names = list_agent_created_skill_names() 207 assert "my-skill" in names 208 assert "bundled-skill" not in names 209 210 211 def test_agent_created_excludes_hub_installed(skills_home): 212 from tools.skill_usage import list_agent_created_skill_names, mark_agent_created 213 skills_dir = skills_home / "skills" 214 _write_skill(skills_dir, "hub-skill") 215 _write_skill(skills_dir, "my-skill") 216 mark_agent_created("my-skill") 217 hub_dir = skills_dir / ".hub" 218 hub_dir.mkdir() 219 (hub_dir / "lock.json").write_text( 220 json.dumps({"version": 1, "installed": {"hub-skill": {"source": "taps/main"}}}), 221 encoding="utf-8", 222 ) 223 names = list_agent_created_skill_names() 224 assert "my-skill" in names 225 assert "hub-skill" not in names 226 227 228 def test_is_agent_created(skills_home): 229 from tools.skill_usage import is_agent_created 230 skills_dir = skills_home / "skills" 231 (skills_dir / ".bundled_manifest").write_text("bundled:abc\n", encoding="utf-8") 232 hub_dir = skills_dir / ".hub" 233 hub_dir.mkdir() 234 (hub_dir / "lock.json").write_text( 235 json.dumps({"installed": {"hubbed": {}}}), encoding="utf-8", 236 ) 237 assert is_agent_created("my-skill") is True 238 assert is_agent_created("bundled") is False 239 assert is_agent_created("hubbed") is False 240 241 242 def test_agent_created_skips_archive_and_hub_dirs(skills_home): 243 from tools.skill_usage import list_agent_created_skill_names, mark_agent_created 244 skills_dir = skills_home / "skills" 245 _write_skill(skills_dir, "real-skill") 246 mark_agent_created("real-skill") 247 # Dot-prefixed dirs must be ignored even if they contain SKILL.md 248 archive = skills_dir / ".archive" / "old-skill" 249 archive.mkdir(parents=True) 250 (archive / "SKILL.md").write_text( 251 "---\nname: old-skill\n---\n", encoding="utf-8", 252 ) 253 names = list_agent_created_skill_names() 254 assert "real-skill" in names 255 assert "old-skill" not in names 256 257 258 # --------------------------------------------------------------------------- 259 # Archive / restore 260 # --------------------------------------------------------------------------- 261 262 def test_archive_skill_moves_directory(skills_home): 263 from tools.skill_usage import archive_skill, get_record, STATE_ARCHIVED 264 skills_dir = skills_home / "skills" 265 skill_dir = _write_skill(skills_dir, "old-skill") 266 assert skill_dir.exists() 267 268 ok, msg = archive_skill("old-skill") 269 assert ok, msg 270 assert not skill_dir.exists() 271 assert (skills_dir / ".archive" / "old-skill" / "SKILL.md").exists() 272 assert get_record("old-skill")["state"] == "archived" 273 assert get_record("old-skill")["archived_at"] is not None 274 275 276 def test_archive_refuses_bundled_skill(skills_home): 277 from tools.skill_usage import archive_skill 278 skills_dir = skills_home / "skills" 279 _write_skill(skills_dir, "bundled") 280 (skills_dir / ".bundled_manifest").write_text("bundled:abc\n", encoding="utf-8") 281 282 ok, msg = archive_skill("bundled") 283 assert not ok 284 assert "bundled" in msg.lower() or "hub" in msg.lower() 285 286 287 def test_archive_refuses_hub_skill(skills_home): 288 from tools.skill_usage import archive_skill 289 skills_dir = skills_home / "skills" 290 _write_skill(skills_dir, "hub-skill") 291 hub_dir = skills_dir / ".hub" 292 hub_dir.mkdir() 293 (hub_dir / "lock.json").write_text( 294 json.dumps({"installed": {"hub-skill": {}}}), encoding="utf-8", 295 ) 296 297 ok, msg = archive_skill("hub-skill") 298 assert not ok 299 300 301 def test_archive_missing_skill_returns_error(skills_home): 302 from tools.skill_usage import archive_skill 303 ok, msg = archive_skill("nonexistent") 304 assert not ok 305 assert "not found" in msg.lower() 306 307 308 def test_restore_skill_moves_back(skills_home): 309 from tools.skill_usage import archive_skill, restore_skill, get_record 310 skills_dir = skills_home / "skills" 311 _write_skill(skills_dir, "temp-skill") 312 archive_skill("temp-skill") 313 assert not (skills_dir / "temp-skill").exists() 314 315 ok, msg = restore_skill("temp-skill") 316 assert ok, msg 317 assert (skills_dir / "temp-skill" / "SKILL.md").exists() 318 assert get_record("temp-skill")["state"] == "active" 319 320 321 def test_restore_skill_finds_nested_archive_subdir(skills_home): 322 """Skills archived under nested category subdirs (e.g. 323 .archive/<category>/<skill>/) — left behind by older archive layouts or 324 external imports — must still be restorable by name.""" 325 from tools.skill_usage import restore_skill, get_record 326 skills_dir = skills_home / "skills" 327 nested = skills_dir / ".archive" / "openclaw-imports" / "nested-skill" 328 nested.mkdir(parents=True) 329 (nested / "SKILL.md").write_text( 330 "---\nname: nested-skill\ndescription: x\n---\n", encoding="utf-8", 331 ) 332 333 ok, msg = restore_skill("nested-skill") 334 assert ok, msg 335 assert (skills_dir / "nested-skill" / "SKILL.md").exists() 336 assert not nested.exists() 337 assert get_record("nested-skill")["state"] == "active" 338 339 340 def test_restore_skill_finds_nested_timestamped_prefix(skills_home): 341 """Prefix-match path (timestamped dupes) must also descend into nested 342 archive subdirs, not just .archive/ top-level.""" 343 from tools.skill_usage import restore_skill 344 skills_dir = skills_home / "skills" 345 nested = skills_dir / ".archive" / "imports" / "dup-skill-20260101000000" 346 nested.mkdir(parents=True) 347 (nested / "SKILL.md").write_text( 348 "---\nname: dup-skill\ndescription: x\n---\n", encoding="utf-8", 349 ) 350 351 ok, msg = restore_skill("dup-skill") 352 assert ok, msg 353 assert (skills_dir / "dup-skill" / "SKILL.md").exists() 354 355 356 def test_archive_collision_gets_suffix(skills_home): 357 from tools.skill_usage import archive_skill 358 skills_dir = skills_home / "skills" 359 _write_skill(skills_dir, "dup") 360 archive_skill("dup") 361 _write_skill(skills_dir, "dup") # recreate 362 ok, msg = archive_skill("dup") 363 assert ok 364 # Two entries under .archive/ — second should have a timestamp suffix 365 archived = sorted(p.name for p in (skills_dir / ".archive").iterdir() if p.is_dir()) 366 assert "dup" in archived 367 assert any(n.startswith("dup-") and n != "dup" for n in archived) 368 369 370 # --------------------------------------------------------------------------- 371 # Reporting 372 # --------------------------------------------------------------------------- 373 374 def test_agent_created_report_includes_marked_skills_with_defaults(skills_home): 375 from tools.skill_usage import agent_created_report, bump_view, mark_agent_created 376 skills_dir = skills_home / "skills" 377 _write_skill(skills_dir, "a") 378 _write_skill(skills_dir, "b") 379 mark_agent_created("a") 380 mark_agent_created("b") 381 bump_view("a") 382 rows = agent_created_report() 383 by_name = {r["name"]: r for r in rows} 384 assert "a" in by_name and "b" in by_name 385 assert by_name["a"]["view_count"] == 1 386 # b has only the provenance marker — activity fields still default. 387 assert by_name["b"]["view_count"] == 0 388 assert by_name["b"]["state"] == "active" 389 390 391 def test_manual_skill_with_usage_is_not_curator_managed(skills_home): 392 from tools.skill_usage import agent_created_report, bump_view, list_agent_created_skill_names 393 skills_dir = skills_home / "skills" 394 _write_skill(skills_dir, "manual-skill") 395 396 bump_view("manual-skill") 397 398 assert "manual-skill" not in list_agent_created_skill_names() 399 assert "manual-skill" not in {r["name"] for r in agent_created_report()} 400 401 402 def test_agent_created_report_excludes_bundled_and_hub(skills_home): 403 from tools.skill_usage import agent_created_report, mark_agent_created 404 skills_dir = skills_home / "skills" 405 _write_skill(skills_dir, "mine") 406 _write_skill(skills_dir, "bundled") 407 _write_skill(skills_dir, "hubbed") 408 mark_agent_created("mine") 409 (skills_dir / ".bundled_manifest").write_text("bundled:abc\n", encoding="utf-8") 410 hub = skills_dir / ".hub" 411 hub.mkdir() 412 (hub / "lock.json").write_text( 413 json.dumps({"installed": {"hubbed": {}}}), encoding="utf-8", 414 ) 415 names = {r["name"] for r in agent_created_report()} 416 assert "mine" in names 417 assert "bundled" not in names 418 assert "hubbed" not in names 419 420 421 def test_agent_created_report_derives_activity_from_view_and_patch(skills_home, monkeypatch): 422 import tools.skill_usage as skill_usage 423 424 skills_dir = skills_home / "skills" 425 _write_skill(skills_dir, "mine") 426 timestamps = iter([ 427 "2026-04-30T10:00:00+00:00", 428 "2026-04-30T11:00:00+00:00", 429 "2026-04-30T12:00:00+00:00", 430 "2026-04-30T13:00:00+00:00", 431 ]) 432 monkeypatch.setattr(skill_usage, "_now_iso", lambda: next(timestamps)) 433 434 skill_usage.mark_agent_created("mine") 435 skill_usage.bump_view("mine") 436 skill_usage.bump_patch("mine") 437 438 row = next(r for r in skill_usage.agent_created_report() if r["name"] == "mine") 439 assert row["activity_count"] == 2 440 assert row["last_activity_at"] == "2026-04-30T12:00:00+00:00" 441 442 443 # --------------------------------------------------------------------------- 444 # Provenance guard — telemetry must not leak records for bundled/hub skills 445 # --------------------------------------------------------------------------- 446 447 def test_bump_view_no_op_for_bundled_skill(skills_home): 448 """Telemetry bumps on bundled skills are dropped — the sidecar must stay 449 focused on agent-created skills only.""" 450 from tools.skill_usage import bump_view, load_usage 451 skills_dir = skills_home / "skills" 452 (skills_dir / ".bundled_manifest").write_text( 453 "ship-bundled:abc\n", encoding="utf-8", 454 ) 455 456 bump_view("ship-bundled") 457 assert "ship-bundled" not in load_usage(), ( 458 "bundled skill leaked into .usage.json" 459 ) 460 461 462 def test_bump_patch_no_op_for_hub_skill(skills_home): 463 from tools.skill_usage import bump_patch, load_usage 464 skills_dir = skills_home / "skills" 465 hub = skills_dir / ".hub" 466 hub.mkdir() 467 (hub / "lock.json").write_text( 468 json.dumps({"installed": {"from-hub": {}}}), encoding="utf-8", 469 ) 470 471 bump_patch("from-hub") 472 assert "from-hub" not in load_usage() 473 474 475 def test_bump_use_no_op_for_hub_skill(skills_home): 476 from tools.skill_usage import bump_use, load_usage 477 skills_dir = skills_home / "skills" 478 hub = skills_dir / ".hub" 479 hub.mkdir() 480 (hub / "lock.json").write_text( 481 json.dumps({"installed": {"from-hub": {}}}), encoding="utf-8", 482 ) 483 484 bump_use("from-hub") 485 assert "from-hub" not in load_usage() 486 487 488 def test_set_state_no_op_for_bundled_skill(skills_home): 489 """State transitions on bundled skills must not land in the sidecar.""" 490 from tools.skill_usage import set_state, load_usage, STATE_ARCHIVED 491 skills_dir = skills_home / "skills" 492 (skills_dir / ".bundled_manifest").write_text( 493 "locked:abc\n", encoding="utf-8", 494 ) 495 set_state("locked", STATE_ARCHIVED) 496 assert "locked" not in load_usage() 497 498 499 def test_restore_refuses_to_shadow_bundled_skill(skills_home): 500 """If a bundled skill now occupies the name, refuse to restore.""" 501 from tools.skill_usage import archive_skill, restore_skill 502 skills_dir = skills_home / "skills" 503 _write_skill(skills_dir, "shared-name") 504 archive_skill("shared-name") 505 506 # Now a bundled skill appears with the same name 507 (skills_dir / ".bundled_manifest").write_text( 508 "shared-name:abc\n", encoding="utf-8", 509 ) 510 _write_skill(skills_dir, "shared-name") # bundled install landed 511 512 ok, msg = restore_skill("shared-name") 513 assert not ok 514 assert "bundled" in msg.lower() or "shadow" in msg.lower() 515 516 517 def test_end_to_end_no_code_path_mutates_bundled_skill(skills_home): 518 """The combined guarantee: no curator code path can archive, mark stale, 519 set-state, or persist telemetry for a bundled or hub-installed skill.""" 520 from tools.skill_usage import ( 521 bump_view, bump_use, bump_patch, set_state, set_pinned, 522 archive_skill, load_usage, STATE_STALE, STATE_ARCHIVED, 523 ) 524 skills_dir = skills_home / "skills" 525 _write_skill(skills_dir, "bundled-one") 526 _write_skill(skills_dir, "hub-one") 527 _write_skill(skills_dir, "mine") 528 529 (skills_dir / ".bundled_manifest").write_text( 530 "bundled-one:abc\n", encoding="utf-8", 531 ) 532 hub = skills_dir / ".hub" 533 hub.mkdir() 534 (hub / "lock.json").write_text( 535 json.dumps({"installed": {"hub-one": {}}}), encoding="utf-8", 536 ) 537 538 # Hammer every mutator at the bundled/hub names 539 for name in ("bundled-one", "hub-one"): 540 bump_view(name) 541 bump_use(name) 542 bump_patch(name) 543 set_state(name, STATE_STALE) 544 set_state(name, STATE_ARCHIVED) 545 set_pinned(name, True) 546 ok, _msg = archive_skill(name) 547 assert not ok, f"archive_skill(\"{name}\") should refuse" 548 549 # Sidecar must be clean of all three 550 data = load_usage() 551 assert "bundled-one" not in data 552 assert "hub-one" not in data 553 554 # Directories must still be in place on disk 555 assert (skills_dir / "bundled-one" / "SKILL.md").exists() 556 assert (skills_dir / "hub-one" / "SKILL.md").exists() 557 558 # The agent-created skill can still be mutated normally 559 bump_view("mine") 560 assert load_usage()["mine"]["view_count"] == 1