Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b6104abedc |
@@ -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.
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
91
tests/test_skill_autoloading.py
Normal file
91
tests/test_skill_autoloading.py
Normal 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)
|
||||
Reference in New Issue
Block a user