Dynamic Skill Hot-Reloading for Exo

skill.md changes
  -> Watcher detects (filesystem events or polling)
  -> yields list[SkillChangeEvent] batch
  -> SkillSyncManager receives batch
  -> Updates SkillRegistry._skills dict
  -> Calls ToolResolver to map skills to Tool objects
  -> Calls agent.add_tool()/remove_tool() for each bound agent
  -> Optionally rebuilds agent.instructions from active skills
  -> Emits events onto EventBus ("skill:added", "skill:modified", "skill:removed")
  -> Agent picks up tool changes on next step (schemas re-enumerated each step)

packages/exo-core/src/exo/
    skills.py              <-- add: SkillChangeEvent, SkillWatcher ABC, ToolResolver Protocol,
                                    DictToolResolver, SkillSyncManager

packages/exo-skills/                   <-- NEW PACKAGE
    pyproject.toml                     <-- depends on exo-core + watchfiles
    src/exo_skills/
        __init__.py                    <-- re-exports LocalFileWatcher, GitHubPollingWatcher
        watchers/
            __init__.py
            local.py                   <-- LocalFileWatcher (watchfiles-backed)
            github.py                  <-- GitHubPollingWatcher (git pull polling)

from dataclasses import dataclass
from typing import Literal

@dataclass(frozen=True)
class SkillChangeEvent:
    """Represents a single skill change detected by a watcher."""
    kind: Literal["added", "modified", "removed"]
    skill_name: str
    skill: Skill | None          # None only for kind="removed"
    source_path: str             # source directory this came from

from abc import ABC, abstractmethod
from collections.abc import AsyncIterator

class SkillWatcher(ABC):
    """Base class for skill source watchers.

    Implementations monitor a skill source (local directory, GitHub repo,
    etc.) and yield batches of change events. Each batch represents one
    "settle" cycle — filesystem events are debounced, polling results are
    diffed.
    """

    @abstractmethod
    async def watch(self) -> AsyncIterator[list[SkillChangeEvent]]:
        """Yield batches of change events. Blocks between batches."""
        ...

    @abstractmethod
    async def stop(self) -> None:
        """Signal the watcher to shut down. Must cause watch() to return."""
        ...

from typing import Protocol, runtime_checkable

@runtime_checkable
class ToolResolver(Protocol):
    """Maps skill metadata to actual Tool implementations."""
    def resolve(self, skill: Skill) -> list[Tool]: ...

class DictToolResolver:
    """Simple resolver that maps skill names to Tool objects via a dict."""

    def __init__(self, tool_map: dict[str, Tool | list[Tool]]) -> None:
        self._map = tool_map

    def resolve(self, skill: Skill) -> list[Tool]:
        entry = self._map.get(skill.name)
        if entry is None:
            return []
        if isinstance(entry, list):
            return entry
        return [entry]

class SkillSyncManager:
    """Orchestrates skill watchers and pushes changes to bound agents.

    Args:
        registry: The SkillRegistry whose state is kept in sync.
        tool_resolver: Maps skills to Tool objects. Accepts a ToolResolver
            instance or a plain dict[str, Tool | list[Tool]].
        event_bus: Optional EventBus for emitting change events.
        instructions_builder: Optional callable that rebuilds agent
            instructions from the current list of active skills.
    """

    def __init__(
        self,
        registry: SkillRegistry,
        tool_resolver: ToolResolver | dict[str, Tool | list[Tool]],
        *,
        event_bus: EventBus | None = None,
        instructions_builder: Callable[[list[Skill]], str] | None = None,
    ) -> None: ...

    def add_watcher(self, watcher: SkillWatcher) -> None:
        """Register a watcher. Must be called before start()."""

    def bind_agent(self, agent: Agent) -> None:
        """Bind an agent to receive skill change updates."""

    def unbind_agent(self, agent: Agent) -> None:
        """Stop pushing updates to an agent."""

    async def start(self) -> None:
        """Start all watchers as background asyncio.Tasks."""

    async def stop(self) -> None:
        """Stop all watchers, cancel tasks, await cleanup."""

    async def __aenter__(self) -> SkillSyncManager: ...
    async def __aexit__(self, *exc) -> None: ...

class LocalFileWatcher(SkillWatcher):
    """Watches a local directory for skill.md changes using watchfiles.

    Args:
        path: Directory to watch (recursive).
        debounce_ms: Milliseconds to wait for filesystem to settle.
    """

    def __init__(self, path: str | Path, debounce_ms: int = 500) -> None: ...

class GitHubPollingWatcher(SkillWatcher):
    """Polls a GitHub repository for skill changes via periodic git pull.

    Args:
        source_url: GitHub URL (parsed via parse_github_url).
        poll_interval: Seconds between polls (default 300 = 5 min).
        cache_dir: Override for clone cache directory.
    """

    def __init__(
        self,
        source_url: str,
        poll_interval: float = 300.0,
        cache_dir: Path | None = None,
    ) -> None: ...

from exo import Agent, run, tool
from exo.skills import SkillRegistry, SkillSyncManager
from exo_skills.watchers import LocalFileWatcher

@tool
async def web_search(query: str) -> str:
    """Search the web."""
    return f"Results for: {query}"

registry = SkillRegistry()
registry.register_source("./skills")
registry.load_all()

agent = Agent(name="assistant", model="openai:gpt-4o", tools=[web_search])

async with SkillSyncManager(
    registry=registry,
    tool_resolver={"web_search": web_search},
) as manager:
    manager.add_watcher(LocalFileWatcher("./skills"))
    manager.bind_agent(agent)
    await manager.start()

    # Agent now auto-picks up skill changes
    result = await run(agent, "Search for AI news")

from exo.events import EventBus

bus = EventBus()

async def log_skill_change(**data):
    print(f"Skill change: {data}")

bus.on("skill:added", log_skill_change)
bus.on("skill:modified", log_skill_change)
bus.on("skill:removed", log_skill_change)

async with SkillSyncManager(
    registry=registry,
    tool_resolver=tool_map,
    event_bus=bus,
) as manager:
    ...

def build_instructions(skills: list[Skill]) -> str:
    parts = ["You have these capabilities:\n"]
    for s in skills:
        parts.append(f"- **{s.name}**: {s.description}")
        if s.usage:
            parts.append(f"  {s.usage.splitlines()[0]}")
    return "\n".join(parts)

async with SkillSyncManager(
    registry=registry,
    tool_resolver=tool_map,
    instructions_builder=build_instructions,
) as manager:
    ...

from exo_skills.watchers import LocalFileWatcher, GitHubPollingWatcher

async with SkillSyncManager(registry=registry, tool_resolver=tool_map) as manager:
    manager.add_watcher(LocalFileWatcher("./local-skills"))
    manager.add_watcher(GitHubPollingWatcher(
        "https://github.com/org/shared-skills/tree/main/skills",
        poll_interval=600,  # 10 minutes
    ))
    manager.bind_agent(agent)
    await manager.start()