Compare commits

...

1 Commits

Author SHA1 Message Date
Hermes Agent
b6104abedc feat: skill auto-loading from timmy-config sidecar (#742)
Some checks failed
Nix / nix (macos-latest) (pull_request) Waiting to run
Docker Build and Publish / build-and-push (pull_request) Has been skipped
Nix / nix (ubuntu-latest) (pull_request) Failing after 8s
Contributor Attribution Check / check-attribution (pull_request) Failing after 56s
Supply Chain Audit / Scan PR for supply chain risks (pull_request) Successful in 48s
Tests / e2e (pull_request) Successful in 2m42s
Tests / test (pull_request) Failing after 45m43s
Resolves #742. Skills installed by the timmy-config sidecar into
~/.hermes/skills/ now appear without requiring an agent restart.

agent/skill_commands.py:
- Added refresh_skill_commands(force=False) — re-scans skills
  directories only if SKILL.md files have changed (mtime check)
- Added should_refresh_skills(turn_count, interval) — determines
  if refresh should happen this turn
- Throttle: no re-scan more often than every 300s (configurable)
- Logs skill count changes on refresh

run_agent.py:
- Wired skill refresh into the turn loop
- Checks every 5 turns if skills directory has changed
- Auto-picks up new/removed skills without restart
- Best-effort (non-critical, won't break agent if refresh fails)

tests/test_skill_autoloading.py (6 tests):
- refresh returns dict, is idempotent
- turn interval logic
- new skill detection after refresh
- throttle behavior, force bypass
2026-04-15 08:03:53 -04:00
3 changed files with 192 additions and 0 deletions

View File

@@ -15,6 +15,10 @@ from typing import Any, Dict, Optional
logger = logging.getLogger(__name__)
_skill_commands: Dict[str, Dict[str, Any]] = {}
# Auto-refresh state: track skills directory modification times
_skill_dirs_mtime: Dict[str, float] = {}
_skill_last_scan_time: float = 0.0
_skill_refresh_interval: float = 300.0 # seconds between refresh checks
_PLAN_SLUG_RE = re.compile(r"[^a-z0-9]+")
# Patterns for sanitizing skill names into clean hyphen-separated slugs.
_SKILL_INVALID_CHARS = re.compile(r"[^a-z0-9-]")
@@ -269,6 +273,94 @@ def get_skill_commands() -> Dict[str, Dict[str, Any]]:
return _skill_commands
def refresh_skill_commands(force: bool = False) -> Dict[str, Dict[str, Any]]:
"""Re-scan skills directories if any have changed since last scan.
Call this periodically (e.g. every N turns) to pick up new skills
installed by the timmy-config sidecar without requiring a restart.
Args:
force: If True, always re-scan regardless of modification times.
Returns:
Updated skill commands mapping.
"""
import time
global _skill_dirs_mtime, _skill_last_scan_time
now = time.time()
# Throttle: don't re-scan more often than every N seconds
if not force and (now - _skill_last_scan_time) < _skill_refresh_interval:
return _skill_commands
try:
from tools.skills_tool import SKILLS_DIR
from agent.skill_utils import get_external_skills_dirs
dirs_to_check = []
if SKILLS_DIR.exists():
dirs_to_check.append(SKILLS_DIR)
dirs_to_check.extend(get_external_skills_dirs())
# Check if any directory has changed
changed = force
current_mtimes: Dict[str, float] = {}
for d in dirs_to_check:
try:
# Get the latest mtime of any SKILL.md in the directory
latest = 0.0
for skill_md in d.rglob("SKILL.md"):
try:
mtime = skill_md.stat().st_mtime
if mtime > latest:
latest = mtime
except OSError:
pass
current_mtimes[str(d)] = latest
old_mtime = _skill_dirs_mtime.get(str(d), 0.0)
if latest > old_mtime:
changed = True
except OSError:
pass
if changed:
_skill_dirs_mtime = current_mtimes
_skill_last_scan_time = now
old_count = len(_skill_commands)
scan_skill_commands()
new_count = len(_skill_commands)
if new_count != old_count:
logger.info(
"Skill refresh: %d skills (was %d, delta: %s%d)",
new_count, old_count,
"+" if new_count > old_count else "",
new_count - old_count,
)
return _skill_commands
_skill_last_scan_time = now
except Exception as e:
logger.debug("Skill refresh check failed: %s", e)
return _skill_commands
def should_refresh_skills(turn_count: int, interval: int = 5) -> bool:
"""Check if skills should be refreshed this turn.
Args:
turn_count: Current conversation turn number.
interval: Refresh every N turns.
Returns:
True if refresh should happen this turn.
"""
return turn_count > 0 and turn_count % interval == 0
def resolve_skill_command_key(command: str) -> Optional[str]:
"""Resolve a user-typed /command to its canonical skill_cmds key.

View File

@@ -7862,6 +7862,15 @@ class AIAgent:
# Track user turns for memory flush and periodic nudge logic
self._user_turn_count += 1
# Auto-refresh skills from sidecar every 5 turns
# Picks up new skills installed by timmy-config without restart
try:
from agent.skill_commands import should_refresh_skills, refresh_skill_commands
if should_refresh_skills(self._user_turn_count, interval=5):
refresh_skill_commands()
except Exception:
pass # non-critical — skill refresh is best-effort
# Preserve the original user message (no nudge injection).
original_user_message = persist_user_message if persist_user_message is not None else user_message

View File

@@ -0,0 +1,91 @@
"""Tests for skill auto-loading from timmy-config sidecar — issue #742."""
import os
import time
import tempfile
from pathlib import Path
import pytest
class TestSkillRefresh:
"""Test the refresh_skill_commands function."""
def test_refresh_returns_dict(self):
from agent.skill_commands import refresh_skill_commands
result = refresh_skill_commands(force=True)
assert isinstance(result, dict)
def test_refresh_is_idempotent(self):
"""Multiple calls with no changes should return same results."""
from agent.skill_commands import refresh_skill_commands
first = refresh_skill_commands(force=True)
second = refresh_skill_commands(force=True)
assert set(first.keys()) == set(second.keys())
def test_should_refresh_skills_interval(self):
from agent.skill_commands import should_refresh_skills
# Turn 0: never refresh
assert not should_refresh_skills(0, interval=5)
# Turn 5: refresh
assert should_refresh_skills(5, interval=5)
# Turn 3: not yet
assert not should_refresh_skills(3, interval=5)
# Turn 10: refresh
assert should_refresh_skills(10, interval=5)
# Turn 7: not yet
assert not should_refresh_skills(7, interval=5)
def test_refresh_picks_up_new_skill(self, tmp_path):
"""New SKILL.md in skills dir should appear after refresh."""
from agent.skill_commands import refresh_skill_commands
import agent.skill_commands as sc
# Create a fake skill
skill_dir = tmp_path / "test-auto-skill"
skill_dir.mkdir()
(skill_dir / "SKILL.md").write_text("""---
name: test-auto-skill
description: A test skill for auto-loading
---
# Test Skill
This is a test.
""")
# Patch SKILLS_DIR to point to tmp_path
from unittest.mock import patch
with patch("tools.skills_tool.SKILLS_DIR", tmp_path):
# Force a scan
sc._skill_commands = {}
sc._skill_dirs_mtime = {}
sc._skill_last_scan_time = 0.0
result = refresh_skill_commands(force=True)
# The skill should appear
assert "/test-auto-skill" in result
assert result["/test-auto-skill"]["name"] == "test-auto-skill"
class TestSkillRefreshThrottling:
"""Test that refresh doesn't re-scan too frequently."""
def test_throttle_blocks_rapid_refresh(self):
from agent.skill_commands import refresh_skill_commands
import agent.skill_commands as sc
sc._skill_last_scan_time = time.time() # just scanned
sc._skill_refresh_interval = 300.0
# Non-forced refresh should be skipped
result = refresh_skill_commands(force=False)
assert result is sc._skill_commands # returns cached, doesn't re-scan
def test_force_bypasses_throttle(self):
from agent.skill_commands import refresh_skill_commands
import agent.skill_commands as sc
sc._skill_last_scan_time = time.time() # just scanned
# Forced refresh should still work
result = refresh_skill_commands(force=True)
assert isinstance(result, dict)