diff --git a/src/strands/vended_plugins/skills/agent_skills.py b/src/strands/vended_plugins/skills/agent_skills.py index ded2afb79..0d21cc2d1 100644 --- a/src/strands/vended_plugins/skills/agent_skills.py +++ b/src/strands/vended_plugins/skills/agent_skills.py @@ -3,11 +3,20 @@ This module provides the AgentSkills class that extends the Plugin base class to add Agent Skills support. The plugin registers a tool for activating skills, and injects skill metadata into the system prompt. + +Sandbox skill sources (e.g., ``"sandbox:///home/skills"``) are loaded +asynchronously during ``init_agent()`` using the agent's sandbox instance, +following the same deferred-loading pattern as the TypeScript SDK. + +The skill catalog is maintained **per-agent** using a ``WeakKeyDictionary``, +so a single plugin instance can be safely shared across multiple agents +(even with different sandboxes) without skill cross-contamination. """ from __future__ import annotations import logging +import weakref from pathlib import Path from typing import TYPE_CHECKING, Any, TypeAlias from xml.sax.saxutils import escape @@ -27,9 +36,10 @@ _DEFAULT_STATE_KEY = "agent_skills" _RESOURCE_DIRS = ("scripts", "references", "assets") _DEFAULT_MAX_RESOURCE_FILES = 20 +_SANDBOX_PREFIX = "sandbox://" SkillSource: TypeAlias = str | Path | Skill -"""A single skill source: path string, Path object, or Skill instance.""" +"""A single skill source: path string, Path object, Skill instance, or ``"sandbox:///path"`` string.""" SkillSources: TypeAlias = SkillSource | list[SkillSource] """One or more skill sources.""" @@ -52,7 +62,16 @@ class AgentSkills(Plugin): 3. Session persistence of active skill state via ``agent.state`` Skills can be provided as filesystem paths (to individual skill directories or - parent directories containing multiple skills) or as pre-built ``Skill`` instances. + parent directories containing multiple skills), ``"sandbox:///path"`` URIs that + load skills from the agent's sandbox at init time, or pre-built ``Skill`` instances. + + Sandbox sources are loaded asynchronously during ``init_agent()`` using the + agent's sandbox, following the same deferred-loading pattern as the TypeScript SDK. + + The skill catalog is stored **per-agent** so that a single plugin instance + can be safely attached to multiple agents with different sandboxes. + Synchronous sources (filesystem, URL, Skill instances) are shared as the + base set; sandbox-resolved skills are added per-agent on top of that base. Example: ```python @@ -62,11 +81,21 @@ class AgentSkills(Plugin): # Load from filesystem plugin = AgentSkills(skills=["./skills/pdf-processing", "./skills/"]) + # Load from agent's sandbox (resolved at agent init) + plugin = AgentSkills(skills=["sandbox:///home/skills"]) + + # Mix local and sandbox sources + plugin = AgentSkills(skills=["./local-skills/", "sandbox:///home/skills"]) + # Or provide Skill instances directly skill = Skill(name="my-skill", description="A custom skill", instructions="Do the thing") plugin = AgentSkills(skills=[skill]) agent = Agent(plugins=[plugin]) + + # Safe to share across multiple agents with different sandboxes + agent_a = Agent(sandbox=sandbox_a, plugins=[plugin]) + agent_b = Agent(sandbox=sandbox_b, plugins=[plugin]) ``` """ @@ -81,6 +110,11 @@ def __init__( ) -> None: """Initialize the AgentSkills plugin. + Synchronous sources (filesystem paths, ``Skill`` instances, HTTPS URLs) + are resolved immediately into the base skill set. Sandbox sources + (``"sandbox:///path"``) are stored as pending and resolved per-agent + during ``init_agent()`` when each agent's sandbox is available. + Args: skills: One or more skill sources. Can be a single value or a list. Each element can be: @@ -88,27 +122,136 @@ def __init__( - A ``str`` or ``Path`` to a parent directory (containing skill subdirectories) - A ``Skill`` dataclass instance - An ``https://`` URL pointing directly to raw SKILL.md content + - A ``"sandbox:///path"`` URI to load from the agent's sandbox at init time state_key: Key used to store plugin state in ``agent.state``. max_resource_files: Maximum number of resource files to list in skill responses. strict: If True, raise on skill validation issues. If False (default), warn and load anyway. """ self._strict = strict - self._skills: dict[str, Skill] = self._resolve_skills(_normalize_sources(skills)) self._state_key = state_key self._max_resource_files = max_resource_files + + # Split sources into sync and deferred (sandbox) groups + all_sources = _normalize_sources(skills) + sync_sources: list[SkillSource] = [] + self._sandbox_sources: list[str] = [] + + for source in all_sources: + if isinstance(source, str) and source.startswith(_SANDBOX_PREFIX): + self._sandbox_sources.append(source[len(_SANDBOX_PREFIX) :]) + else: + sync_sources.append(source) + + # Resolve synchronous sources immediately — shared across all agents + self._base_skills: dict[str, Skill] = self._resolve_skills(sync_sources) + + # Per-agent skill catalogs (base + sandbox-resolved skills) + # Uses WeakKeyDictionary so entries are cleaned up when agents are GC'd + self._agent_skills: weakref.WeakKeyDictionary[Agent, dict[str, Skill]] = weakref.WeakKeyDictionary() + super().__init__() - def init_agent(self, agent: Agent) -> None: + def _get_skills(self, agent: Agent | None = None) -> dict[str, Skill]: + """Get the skill catalog for a specific agent. + + Returns the per-agent catalog if the agent has been initialized, + otherwise falls back to the base (sync-resolved) skills. + + Args: + agent: The agent to get skills for. If None, returns base skills. + + Returns: + Dict mapping skill names to Skill instances. + """ + if agent is not None and agent in self._agent_skills: + return self._agent_skills[agent] + if agent is not None and agent not in self._agent_skills and self._sandbox_sources: + logger.warning( + "agent has not been initialized yet — sandbox skills are not available; returning base skills only" + ) + return self._base_skills + + async def init_agent(self, agent: Agent) -> None: """Initialize the plugin with an agent instance. - Decorated hooks and tools are auto-registered by the plugin registry. + Creates a per-agent skill catalog by copying the base skills and + then resolving any sandbox sources using the agent's sandbox. This + ensures each agent gets its own skill set without cross-contamination. Args: agent: The agent instance to extend with skills support. """ - if not self._skills: + # Start with a COPY of base skills for this agent + agent_skills = dict(self._base_skills) + + # Resolve deferred sandbox sources from THIS agent's sandbox + if self._sandbox_sources: + sandbox_skills = await self._resolve_sandbox_skills(agent) + agent_skills.update(sandbox_skills) + + # Store per-agent catalog + self._agent_skills[agent] = agent_skills + + if not agent_skills: logger.warning("no skills were loaded, the agent will have no skills available") - logger.debug("skill_count=<%d> | skills plugin initialized", len(self._skills)) + logger.debug("skill_count=<%d> | skills plugin initialized for agent", len(agent_skills)) + + async def _resolve_sandbox_skills(self, agent: Agent) -> dict[str, Skill]: + """Resolve sandbox skill sources using the agent's sandbox. + + Each sandbox source path is treated as either a single skill directory + (if it contains SKILL.md) or a parent directory containing skill + subdirectories. + + Args: + agent: The agent whose sandbox to load skills from. + + Returns: + Dict mapping skill names to Skill instances loaded from the sandbox. + """ + resolved: dict[str, Skill] = {} + + sandbox = getattr(agent, "sandbox", None) + if sandbox is None: + logger.warning( + "agent has no sandbox configured — skipping %d sandbox skill source(s)", + len(self._sandbox_sources), + ) + return resolved + + for sandbox_path in self._sandbox_sources: + logger.debug("sandbox_path=<%s> | resolving sandbox skill source", sandbox_path) + + try: + # First try loading as a single skill + skill = await Skill.from_sandbox(sandbox, sandbox_path, strict=self._strict) + if skill.name in resolved: + logger.warning("name=<%s> | duplicate skill name, overwriting previous skill", skill.name) + resolved[skill.name] = skill + logger.debug( + "sandbox_path=<%s>, name=<%s> | loaded single skill from sandbox", + sandbox_path, + skill.name, + ) + except FileNotFoundError: + # Not a single skill — try as parent directory + try: + skills = await Skill.from_sandbox_directory(sandbox, sandbox_path, strict=self._strict) + for skill in skills: + if skill.name in resolved: + logger.warning("name=<%s> | duplicate skill name, overwriting previous skill", skill.name) + resolved[skill.name] = skill + logger.debug( + "sandbox_path=<%s>, count=<%d> | loaded skills from sandbox directory", + sandbox_path, + len(skills), + ) + except Exception as e: + logger.warning("sandbox_path=<%s> | failed to load sandbox skills: %s", sandbox_path, e) + except Exception as e: + logger.warning("sandbox_path=<%s> | failed to load sandbox skill: %s", sandbox_path, e) + + return resolved @tool(context=True) def skills(self, skill_name: str, tool_context: ToolContext) -> str: # noqa: D417 @@ -120,13 +263,15 @@ def skills(self, skill_name: str, tool_context: ToolContext) -> str: # noqa: D4 Args: skill_name: Name of the skill to activate. """ + agent_skills = self._get_skills(tool_context.agent) + if not skill_name: - available = ", ".join(self._skills) + available = ", ".join(agent_skills) return f"Error: skill_name is required. Available skills: {available}" - found = self._skills.get(skill_name) + found = agent_skills.get(skill_name) if found is None: - available = ", ".join(self._skills) + available = ", ".join(agent_skills) return f"Skill '{skill_name}' not found. Available skills: {available}" logger.debug("skill_name=<%s> | skill activated", skill_name) @@ -154,7 +299,7 @@ def _on_before_invocation(self, event: BeforeInvocationEvent) -> None: state_data = agent.state.get(self._state_key) last_injected_xml = state_data.get("last_injected_xml") if isinstance(state_data, dict) else None - skills_xml = self._generate_skills_xml() + skills_xml = self._generate_skills_xml(agent) content = agent.system_prompt_content if content is not None: @@ -183,13 +328,20 @@ def _on_before_invocation(self, event: BeforeInvocationEvent) -> None: self._set_state_field(agent, "last_injected_xml", new_injected_xml) agent.system_prompt = new_prompt - def get_available_skills(self) -> list[Skill]: + def get_available_skills(self, agent: Agent | None = None) -> list[Skill]: """Get the list of available skills. + When called with an agent, returns that agent's full skill catalog + (base + sandbox-resolved). Without an agent, returns only the base + (sync-resolved) skills. + + Args: + agent: Optional agent to get per-agent skills for. + Returns: - A copy of the current skills list. + A copy of the skills list. """ - return list(self._skills.values()) + return list(self._get_skills(agent).values()) def set_available_skills(self, skills: SkillSources) -> None: """Set the available skills, replacing any existing ones. @@ -199,14 +351,31 @@ def set_available_skills(self, skills: SkillSources) -> None: parent directory containing skill subdirectories, or an ``https://`` URL pointing directly to raw SKILL.md content. - Note: this does not persist state or deactivate skills on any agent. - Active skill state is managed per-agent and will be reconciled on the - next tool call or invocation. + Note: Sandbox sources (``"sandbox:///path"``) are NOT supported in + ``set_available_skills`` because no agent context is available. + Use ``"sandbox:..."`` sources in the constructor instead. + + Note: this replaces the base skill set and clears all per-agent + caches so they will be rebuilt on the next ``init_agent``. Args: skills: One or more skill sources to resolve and set. + + Raises: + ValueError: If a sandbox source is passed (not supported here). """ - self._skills = self._resolve_skills(_normalize_sources(skills)) + sources = _normalize_sources(skills) + for source in sources: + if isinstance(source, str) and source.startswith(_SANDBOX_PREFIX): + raise ValueError( + f"Sandbox sources ('{source}') are not supported in set_available_skills(). " + "Use sandbox sources in the AgentSkills constructor instead." + ) + + self._base_skills = self._resolve_skills(sources) + self._sandbox_sources = [] + # Clear per-agent caches so they pick up new base skills + self._agent_skills.clear() def _format_skill_response(self, skill: Skill) -> str: """Format the tool response when a skill is activated. @@ -274,22 +443,27 @@ def _list_skill_resources(self, skill_path: Path) -> list[str]: return files - def _generate_skills_xml(self) -> str: + def _generate_skills_xml(self, agent: Agent | None = None) -> str: """Generate the XML block listing available skills for the system prompt. When no skills are loaded, returns a block indicating no skills are available. Otherwise includes a ```` element for skills loaded from the filesystem, following the AgentSkills.io integration spec. + Args: + agent: Optional agent to generate per-agent skills XML for. + Returns: XML-formatted string with skill metadata. """ - if not self._skills: + skills = self._get_skills(agent) + + if not skills: return "\nNo skills are currently available.\n" lines: list[str] = [""] - for skill in self._skills.values(): + for skill in skills.values(): lines.append("") lines.append(f"{escape(skill.name)}") lines.append(f"{escape(skill.description)}") @@ -307,6 +481,9 @@ def _resolve_skills(self, sources: list[SkillSource]) -> dict[str, Skill]: a path to a parent directory containing multiple skills, or an HTTPS URL pointing to a SKILL.md file. + Note: Sandbox sources should be resolved separately via + ``_resolve_sandbox_skills()``. + Args: sources: List of skill sources to resolve. diff --git a/src/strands/vended_plugins/skills/skill.py b/src/strands/vended_plugins/skills/skill.py index a60c1cd6c..b99a8c8eb 100644 --- a/src/strands/vended_plugins/skills/skill.py +++ b/src/strands/vended_plugins/skills/skill.py @@ -14,10 +14,13 @@ import urllib.request from dataclasses import dataclass, field from pathlib import Path -from typing import Any +from typing import TYPE_CHECKING, Any import yaml +if TYPE_CHECKING: + from ...sandbox.base import Sandbox + logger = logging.getLogger(__name__) _SKILL_NAME_PATTERN = re.compile(r"^[a-z0-9]([a-z0-9-]*[a-z0-9])?$") @@ -422,3 +425,101 @@ def from_directory(cls, skills_dir: str | Path, *, strict: bool = False) -> list logger.debug("path=<%s>, count=<%d> | loaded skills from directory", skills_dir, len(skills)) return skills + + @classmethod + async def from_sandbox(cls, sandbox: Sandbox, skill_path: str, *, strict: bool = False) -> Skill: + """Load a single skill from a sandbox filesystem. + + Reads a SKILL.md file from the sandbox and parses it into a Skill + instance. The sandbox can be any implementation (host, Docker, cloud). + + Example:: + + from strands.sandbox import HostSandbox + + sandbox = HostSandbox() + skill = await Skill.from_sandbox(sandbox, "/home/skills/my-skill") + + Args: + sandbox: The sandbox to read from. + skill_path: Path to the skill directory (containing SKILL.md) or the + SKILL.md file itself within the sandbox. + strict: If True, raise on any validation issue. If False (default), + warn and load anyway. + + Returns: + A Skill instance populated from the sandbox SKILL.md file. + + Raises: + FileNotFoundError: If no SKILL.md is found at the given path. + ValueError: If the skill metadata is invalid. + """ + # Normalize path — try SKILL.md first, then skill.md + if skill_path.lower().endswith("skill.md"): + # Direct path to SKILL.md — read it directly + logger.debug("sandbox_path=<%s> | loading skill from sandbox", skill_path) + content = await sandbox.read_text(skill_path) + skill = cls.from_content(content, strict=strict) + logger.debug("name=<%s>, sandbox_path=<%s> | skill loaded from sandbox", skill.name, skill_path) + return skill + + # Try to find and read SKILL.md in the directory (single I/O per attempt) + for name in ("SKILL.md", "skill.md"): + candidate = f"{skill_path.rstrip('/')}/{name}" + try: + content = await sandbox.read_text(candidate) + logger.debug("sandbox_path=<%s> | loading skill from sandbox", candidate) + skill = cls.from_content(content, strict=strict) + logger.debug("name=<%s>, sandbox_path=<%s> | skill loaded from sandbox", skill.name, candidate) + return skill + except (FileNotFoundError, OSError, NotImplementedError): + continue + + raise FileNotFoundError(f"path=<{skill_path}> | no SKILL.md found in sandbox skill directory") + + @classmethod + async def from_sandbox_directory(cls, sandbox: Sandbox, skills_dir: str, *, strict: bool = False) -> list[Skill]: + """Load all skills from a parent directory in a sandbox. + + Lists subdirectories in the given sandbox path and loads any that + contain a SKILL.md file. + + Example:: + + from strands.sandbox import HostSandbox + + sandbox = HostSandbox() + skills = await Skill.from_sandbox_directory(sandbox, "/home/skills") + + Args: + sandbox: The sandbox to read from. + skills_dir: Path to the parent directory containing skill + subdirectories within the sandbox. + strict: If True, raise on any validation issue. If False (default), + warn and load anyway. + + Returns: + List of Skill instances loaded from the sandbox directory. + """ + skills: list[Skill] = [] + + try: + entries = await sandbox.list_files(skills_dir) + except (FileNotFoundError, OSError, NotImplementedError) as e: + logger.warning("sandbox_path=<%s> | failed to list sandbox directory: %s", skills_dir, e) + return skills + + for entry in sorted(entries, key=lambda e: e.name): + if not entry.is_dir: + continue + + child_path = f"{skills_dir.rstrip('/')}/{entry.name}" + + try: + skill = await cls.from_sandbox(sandbox, child_path, strict=strict) + skills.append(skill) + except (FileNotFoundError, ValueError, OSError, NotImplementedError) as e: + logger.debug("sandbox_path=<%s> | skipping directory: %s", child_path, e) + + logger.debug("sandbox_path=<%s>, count=<%d> | loaded skills from sandbox directory", skills_dir, len(skills)) + return skills diff --git a/tests/strands/vended_plugins/skills/test_agent_skills.py b/tests/strands/vended_plugins/skills/test_agent_skills.py index 03f43ef2c..9311a2585 100644 --- a/tests/strands/vended_plugins/skills/test_agent_skills.py +++ b/tests/strands/vended_plugins/skills/test_agent_skills.py @@ -4,6 +4,8 @@ from pathlib import Path from unittest.mock import MagicMock +import pytest + from strands.hooks.events import BeforeInvocationEvent from strands.hooks.registry import HookRegistry from strands.plugins.registry import _PluginRegistry @@ -66,9 +68,11 @@ def _set_system_prompt(agent: MagicMock, value: str | list | None) -> None: agent._system_prompt = value agent._system_prompt_content = [{"text": value}] elif isinstance(value, list): + # Content-block path: list of SystemContentBlock dicts + agent._system_prompt_content = value + # Derive the string prompt from text blocks text_parts = [block["text"] for block in value if "text" in block] agent._system_prompt = "\n".join(text_parts) if text_parts else None - agent._system_prompt_content = value elif value is None: agent._system_prompt = None agent._system_prompt_content = None @@ -160,12 +164,13 @@ def test_registers_hooks(self): assert agent.hooks.has_callbacks() - def test_does_not_store_agent_reference(self): + @pytest.mark.asyncio + async def test_does_not_store_agent_reference(self): """Test that init_agent does not store the agent on the plugin.""" plugin = AgentSkills(skills=[_make_skill()]) agent = _mock_agent() - plugin.init_agent(agent) + await plugin.init_agent(agent) assert not hasattr(plugin, "_agent") @@ -702,16 +707,16 @@ def test_resolve_skill_instances(self): skill = _make_skill() plugin = AgentSkills(skills=[skill]) - assert len(plugin._skills) == 1 - assert plugin._skills["test-skill"] is skill + assert len(plugin._base_skills) == 1 + assert plugin._base_skills["test-skill"] is skill def test_resolve_skill_directory_path(self, tmp_path): """Test resolving a path to a skill directory.""" _make_skill_dir(tmp_path, "path-skill") plugin = AgentSkills(skills=[tmp_path / "path-skill"]) - assert len(plugin._skills) == 1 - assert "path-skill" in plugin._skills + assert len(plugin._base_skills) == 1 + assert "path-skill" in plugin._base_skills def test_resolve_parent_directory_path(self, tmp_path): """Test resolving a path to a parent directory.""" @@ -719,20 +724,20 @@ def test_resolve_parent_directory_path(self, tmp_path): _make_skill_dir(tmp_path, "child-b") plugin = AgentSkills(skills=[tmp_path]) - assert len(plugin._skills) == 2 + assert len(plugin._base_skills) == 2 def test_resolve_skill_md_file_path(self, tmp_path): """Test resolving a path to a SKILL.md file.""" skill_dir = _make_skill_dir(tmp_path, "file-skill") plugin = AgentSkills(skills=[skill_dir / "SKILL.md"]) - assert len(plugin._skills) == 1 - assert "file-skill" in plugin._skills + assert len(plugin._base_skills) == 1 + assert "file-skill" in plugin._base_skills def test_resolve_nonexistent_path(self, tmp_path): """Test that nonexistent paths are skipped.""" plugin = AgentSkills(skills=[str(tmp_path / "ghost")]) - assert len(plugin._skills) == 0 + assert len(plugin._base_skills) == 0 class TestResolveUrlSkills: diff --git a/tests/strands/vended_plugins/skills/test_sandbox_skills.py b/tests/strands/vended_plugins/skills/test_sandbox_skills.py new file mode 100644 index 000000000..69677e9cb --- /dev/null +++ b/tests/strands/vended_plugins/skills/test_sandbox_skills.py @@ -0,0 +1,852 @@ +"""Tests for sandbox-based skill loading in the AgentSkills plugin. + +Tests cover: +- Skill.from_sandbox() — loading a single skill from sandbox +- Skill.from_sandbox_directory() — loading multiple skills from sandbox +- AgentSkills with "sandbox:///path" sources — deferred loading in init_agent +- Mixed local + sandbox sources +- Error handling (missing files, invalid content, sandbox failures) +- Multi-agent isolation with different sandboxes +""" + +import logging +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from strands.hooks.registry import HookRegistry +from strands.sandbox.base import FileInfo, Sandbox +from strands.types.tools import ToolContext +from strands.vended_plugins.skills.agent_skills import AgentSkills +from strands.vended_plugins.skills.skill import Skill + +# --- Helpers --- + +SKILL_CONTENT = """--- +name: sandbox-skill +description: A skill loaded from sandbox +allowed-tools: shell editor +--- +# Sandbox Skill Instructions + +Follow these steps to do the thing. +""" + +SKILL_B_CONTENT = """--- +name: another-skill +description: Another sandbox skill +--- +# Another Skill + +More instructions. +""" + +SKILL_C_CONTENT = """--- +name: docker-skill +description: A skill from Docker sandbox +--- +# Docker Skill + +Docker-specific instructions. +""" + +SKILL_D_CONTENT = """--- +name: s3-skill +description: A skill from S3 sandbox +--- +# S3 Skill + +S3-specific instructions. +""" + +INVALID_SKILL_CONTENT = """--- +description: Missing name field +--- +# Broken +""" + + +def _make_mock_sandbox(files: dict[str, str | bytes] | None = None, dirs: dict[str, list[FileInfo]] | None = None): + """Create a mock Sandbox with configurable file system. + + Args: + files: Mapping of path -> content (str for text, bytes for binary). + dirs: Mapping of path -> list of FileInfo entries. + """ + files = files or {} + dirs = dirs or {} + + sandbox = AsyncMock(spec=Sandbox) + + async def mock_read_file(path, **kwargs): + if path in files: + content = files[path] + return content.encode("utf-8") if isinstance(content, str) else content + raise FileNotFoundError(f"No such file: {path}") + + async def mock_read_text(path, encoding="utf-8", **kwargs): + if path in files: + content = files[path] + return content if isinstance(content, str) else content.decode(encoding) + raise FileNotFoundError(f"No such file: {path}") + + async def mock_list_files(path, **kwargs): + if path in dirs: + return dirs[path] + raise FileNotFoundError(f"No such directory: {path}") + + sandbox.read_file = AsyncMock(side_effect=mock_read_file) + sandbox.read_text = AsyncMock(side_effect=mock_read_text) + sandbox.list_files = AsyncMock(side_effect=mock_list_files) + + return sandbox + + +def _mock_agent(sandbox=None): + """Create a mock agent with sandbox support.""" + agent = MagicMock() + agent._system_prompt = "You are an agent." + + type(agent).system_prompt = property( + lambda self: self._system_prompt, + lambda self, value: setattr(self, "_system_prompt", value), + ) + + agent.hooks = HookRegistry() + agent.add_hook = MagicMock( + side_effect=lambda callback, event_type=None: agent.hooks.add_callback(event_type, callback) + ) + agent.tool_registry = MagicMock() + agent.tool_registry.process_tools = MagicMock(return_value=["skills"]) + + state_store: dict[str, object] = {} + agent.state = MagicMock() + agent.state.get = MagicMock(side_effect=lambda key: state_store.get(key)) + agent.state.set = MagicMock(side_effect=lambda key, value: state_store.__setitem__(key, value)) + + if sandbox is not None: + agent.sandbox = sandbox + else: + agent.sandbox = _make_mock_sandbox() + + return agent + + +# --- Tests for Skill.from_sandbox --- + + +class TestSkillFromSandbox: + """Tests for Skill.from_sandbox classmethod.""" + + @pytest.mark.asyncio + async def test_load_skill_from_sandbox_directory(self): + """Test loading a skill from a directory containing SKILL.md.""" + sandbox = _make_mock_sandbox(files={"/home/skills/my-skill/SKILL.md": SKILL_CONTENT}) + + skill = await Skill.from_sandbox(sandbox, "/home/skills/my-skill") + + assert skill.name == "sandbox-skill" + assert skill.description == "A skill loaded from sandbox" + assert "Follow these steps" in skill.instructions + assert skill.allowed_tools == ["shell", "editor"] + + @pytest.mark.asyncio + async def test_load_skill_from_direct_skill_md_path(self): + """Test loading a skill by pointing directly to SKILL.md.""" + sandbox = _make_mock_sandbox(files={"/home/skills/my-skill/SKILL.md": SKILL_CONTENT}) + + skill = await Skill.from_sandbox(sandbox, "/home/skills/my-skill/SKILL.md") + + assert skill.name == "sandbox-skill" + + @pytest.mark.asyncio + async def test_load_skill_lowercase_skill_md(self): + """Test loading skill from skill.md (lowercase).""" + sandbox = _make_mock_sandbox(files={"/home/skills/my-skill/skill.md": SKILL_CONTENT}) + + skill = await Skill.from_sandbox(sandbox, "/home/skills/my-skill") + + assert skill.name == "sandbox-skill" + + @pytest.mark.asyncio + async def test_prefers_uppercase_skill_md(self): + """Test that SKILL.md is preferred over skill.md.""" + sandbox = _make_mock_sandbox( + files={ + "/home/skills/my-skill/SKILL.md": SKILL_CONTENT, + "/home/skills/my-skill/skill.md": SKILL_B_CONTENT, + } + ) + + skill = await Skill.from_sandbox(sandbox, "/home/skills/my-skill") + + assert skill.name == "sandbox-skill" # From SKILL.md, not skill.md + + @pytest.mark.asyncio + async def test_raises_when_no_skill_md(self): + """Test FileNotFoundError when directory has no SKILL.md.""" + sandbox = _make_mock_sandbox() + + with pytest.raises(FileNotFoundError, match="no SKILL.md found"): + await Skill.from_sandbox(sandbox, "/home/skills/empty-dir") + + @pytest.mark.asyncio + async def test_raises_on_invalid_content(self): + """Test ValueError when SKILL.md has invalid content.""" + sandbox = _make_mock_sandbox(files={"/home/skills/bad-skill/SKILL.md": INVALID_SKILL_CONTENT}) + + with pytest.raises(ValueError, match="name"): + await Skill.from_sandbox(sandbox, "/home/skills/bad-skill") + + @pytest.mark.asyncio + async def test_path_trailing_slash(self): + """Test that trailing slashes in path are handled correctly.""" + sandbox = _make_mock_sandbox(files={"/home/skills/my-skill/SKILL.md": SKILL_CONTENT}) + + skill = await Skill.from_sandbox(sandbox, "/home/skills/my-skill/") + + assert skill.name == "sandbox-skill" + + @pytest.mark.asyncio + async def test_strict_mode(self): + """Test strict validation mode.""" + content = """--- +name: Bad_Name +description: Has invalid name +--- +# Instructions +""" + sandbox = _make_mock_sandbox(files={"/home/skills/bad/SKILL.md": content}) + + with pytest.raises(ValueError, match="skill name"): + await Skill.from_sandbox(sandbox, "/home/skills/bad", strict=True) + + +class TestSkillFromSandboxDirectory: + """Tests for Skill.from_sandbox_directory classmethod.""" + + @pytest.mark.asyncio + async def test_load_multiple_skills(self): + """Test loading multiple skills from a parent directory.""" + sandbox = _make_mock_sandbox( + files={ + "/home/skills/skill-a/SKILL.md": SKILL_CONTENT, + "/home/skills/skill-b/SKILL.md": SKILL_B_CONTENT, + }, + dirs={ + "/home/skills": [ + FileInfo(name="skill-a", is_dir=True), + FileInfo(name="skill-b", is_dir=True), + ] + }, + ) + + skills = await Skill.from_sandbox_directory(sandbox, "/home/skills") + + assert len(skills) == 2 + names = {s.name for s in skills} + assert "sandbox-skill" in names + assert "another-skill" in names + + @pytest.mark.asyncio + async def test_skips_non_directory_entries(self): + """Test that files in the parent directory are skipped.""" + sandbox = _make_mock_sandbox( + files={"/home/skills/skill-a/SKILL.md": SKILL_CONTENT}, + dirs={ + "/home/skills": [ + FileInfo(name="skill-a", is_dir=True), + FileInfo(name="README.md", is_dir=False), + ] + }, + ) + + skills = await Skill.from_sandbox_directory(sandbox, "/home/skills") + + assert len(skills) == 1 + assert skills[0].name == "sandbox-skill" + + @pytest.mark.asyncio + async def test_skips_directories_without_skill_md(self): + """Test that directories without SKILL.md are silently skipped.""" + sandbox = _make_mock_sandbox( + files={"/home/skills/skill-a/SKILL.md": SKILL_CONTENT}, + dirs={ + "/home/skills": [ + FileInfo(name="skill-a", is_dir=True), + FileInfo(name="empty-dir", is_dir=True), + ] + }, + ) + + skills = await Skill.from_sandbox_directory(sandbox, "/home/skills") + + assert len(skills) == 1 + + @pytest.mark.asyncio + async def test_empty_directory(self): + """Test loading from an empty directory.""" + sandbox = _make_mock_sandbox(dirs={"/home/skills": []}) + + skills = await Skill.from_sandbox_directory(sandbox, "/home/skills") + + assert skills == [] + + @pytest.mark.asyncio + async def test_nonexistent_directory(self): + """Test loading from a directory that doesn't exist.""" + sandbox = _make_mock_sandbox() + + skills = await Skill.from_sandbox_directory(sandbox, "/nonexistent") + + assert skills == [] + + @pytest.mark.asyncio + async def test_skips_invalid_skills_with_warning(self, caplog): + """Test that invalid skills are skipped with a debug log.""" + sandbox = _make_mock_sandbox( + files={ + "/home/skills/good-skill/SKILL.md": SKILL_CONTENT, + "/home/skills/bad-skill/SKILL.md": INVALID_SKILL_CONTENT, + }, + dirs={ + "/home/skills": [ + FileInfo(name="bad-skill", is_dir=True), + FileInfo(name="good-skill", is_dir=True), + ] + }, + ) + + with caplog.at_level(logging.DEBUG): + skills = await Skill.from_sandbox_directory(sandbox, "/home/skills") + + assert len(skills) == 1 + assert skills[0].name == "sandbox-skill" + + @pytest.mark.asyncio + async def test_sorted_by_directory_name(self): + """Test that skills are loaded in sorted directory name order.""" + sandbox = _make_mock_sandbox( + files={ + "/home/skills/z-skill/SKILL.md": SKILL_B_CONTENT, + "/home/skills/a-skill/SKILL.md": SKILL_CONTENT, + }, + dirs={ + "/home/skills": [ + FileInfo(name="z-skill", is_dir=True), + FileInfo(name="a-skill", is_dir=True), + ] + }, + ) + + skills = await Skill.from_sandbox_directory(sandbox, "/home/skills") + + assert len(skills) == 2 + # Loaded in sorted order (a-skill first, z-skill second) + assert skills[0].name == "sandbox-skill" # from a-skill/ + assert skills[1].name == "another-skill" # from z-skill/ + + +# --- Tests for AgentSkills with "sandbox:" sources --- + + +class TestAgentSkillsSandboxSources: + """Tests for AgentSkills plugin with sandbox: URI sources.""" + + @pytest.mark.asyncio + async def test_sandbox_source_parsed_from_constructor(self): + """Test that 'sandbox:' prefixed sources are stored as pending.""" + plugin = AgentSkills(skills=["sandbox:///home/skills"]) + + assert len(plugin._sandbox_sources) == 1 + assert plugin._sandbox_sources[0] == "/home/skills" + assert len(plugin._base_skills) == 0 # Not loaded yet + + @pytest.mark.asyncio + async def test_sandbox_skills_loaded_in_init_agent(self): + """Test that sandbox skills are loaded during init_agent.""" + sandbox = _make_mock_sandbox( + files={"/home/skills/my-skill/SKILL.md": SKILL_CONTENT}, + dirs={ + "/home/skills": [ + FileInfo(name="my-skill", is_dir=True), + ] + }, + ) + agent = _mock_agent(sandbox=sandbox) + plugin = AgentSkills(skills=["sandbox:///home/skills"]) + + assert len(plugin._base_skills) == 0 + + await plugin.init_agent(agent) + + # Per-agent catalog should have the skill + assert len(plugin._agent_skills[agent]) == 1 + assert "sandbox-skill" in plugin._agent_skills[agent] + # Base skills should remain empty (sandbox skills are per-agent) + assert len(plugin._base_skills) == 0 + + @pytest.mark.asyncio + async def test_mixed_local_and_sandbox_sources(self, tmp_path): + """Test mixing local filesystem and sandbox sources.""" + # Create a local skill + local_skill_dir = tmp_path / "local-skill" + local_skill_dir.mkdir() + (local_skill_dir / "SKILL.md").write_text( + "---\nname: local-skill\ndescription: From filesystem\n---\n# Local\n" + ) + + # Create sandbox with a different skill + sandbox = _make_mock_sandbox( + files={"/home/skills/remote-skill/SKILL.md": SKILL_CONTENT}, + dirs={ + "/home/skills": [ + FileInfo(name="remote-skill", is_dir=True), + ] + }, + ) + agent = _mock_agent(sandbox=sandbox) + + plugin = AgentSkills(skills=[str(local_skill_dir), "sandbox:///home/skills"]) + + # Local skill resolved immediately into base + assert "local-skill" in plugin._base_skills + + # After init_agent, agent has both base + sandbox skills + await plugin.init_agent(agent) + + agent_skills = plugin._get_skills(agent) + assert "local-skill" in agent_skills + assert "sandbox-skill" in agent_skills + + @pytest.mark.asyncio + async def test_sandbox_single_skill_directory(self): + """Test sandbox source pointing to a single skill directory (not parent).""" + sandbox = _make_mock_sandbox(files={"/home/skills/my-skill/SKILL.md": SKILL_CONTENT}) + agent = _mock_agent(sandbox=sandbox) + + plugin = AgentSkills(skills=["sandbox:///home/skills/my-skill"]) + await plugin.init_agent(agent) + + assert "sandbox-skill" in plugin._get_skills(agent) + + @pytest.mark.asyncio + async def test_sandbox_source_not_found_warns(self, caplog): + """Test that missing sandbox paths warn and don't crash.""" + sandbox = _make_mock_sandbox() + agent = _mock_agent(sandbox=sandbox) + + plugin = AgentSkills(skills=["sandbox:///nonexistent"]) + + with caplog.at_level(logging.WARNING): + await plugin.init_agent(agent) + + assert len(plugin._get_skills(agent)) == 0 + + @pytest.mark.asyncio + async def test_sandbox_source_duplicate_overwrites(self): + """Test that duplicate skill names from sandbox overwrite earlier ones.""" + sandbox = _make_mock_sandbox(files={"/sandbox/skills/dupe/SKILL.md": SKILL_CONTENT}) + agent = _mock_agent(sandbox=sandbox) + + # Pre-load a skill with the same name into base + existing = Skill(name="sandbox-skill", description="Original", instructions="Old") + plugin = AgentSkills(skills=[existing, "sandbox:///sandbox/skills/dupe"]) + await plugin.init_agent(agent) + + # Sandbox version should overwrite the base version in the per-agent catalog + assert plugin._get_skills(agent)["sandbox-skill"].description == "A skill loaded from sandbox" + + @pytest.mark.asyncio + async def test_multiple_sandbox_sources(self): + """Test multiple sandbox: sources.""" + sandbox = _make_mock_sandbox( + files={ + "/skills-a/s1/SKILL.md": SKILL_CONTENT, + "/skills-b/s2/SKILL.md": SKILL_B_CONTENT, + }, + dirs={ + "/skills-a": [FileInfo(name="s1", is_dir=True)], + "/skills-b": [FileInfo(name="s2", is_dir=True)], + }, + ) + agent = _mock_agent(sandbox=sandbox) + + plugin = AgentSkills(skills=["sandbox:///skills-a", "sandbox:///skills-b"]) + await plugin.init_agent(agent) + + agent_skills = plugin._get_skills(agent) + assert len(agent_skills) == 2 + assert "sandbox-skill" in agent_skills + assert "another-skill" in agent_skills + + @pytest.mark.asyncio + async def test_set_available_skills_rejects_sandbox(self): + """Test that set_available_skills raises on sandbox sources.""" + plugin = AgentSkills(skills=[]) + + with pytest.raises(ValueError, match="Sandbox sources"): + plugin.set_available_skills(["sandbox:///home/skills"]) + + @pytest.mark.asyncio + async def test_sandbox_skills_appear_in_system_prompt(self): + """Test that sandbox-loaded skills appear in the system prompt XML.""" + sandbox = _make_mock_sandbox( + files={"/home/skills/my-skill/SKILL.md": SKILL_CONTENT}, + dirs={ + "/home/skills": [FileInfo(name="my-skill", is_dir=True)], + }, + ) + agent = _mock_agent(sandbox=sandbox) + + plugin = AgentSkills(skills=["sandbox:///home/skills"]) + await plugin.init_agent(agent) + + # Simulate before_invocation hook — generate XML for this agent + xml = plugin._generate_skills_xml(agent) + + assert "sandbox-skill" in xml + assert "A skill loaded from sandbox" in xml + + @pytest.mark.asyncio + async def test_sandbox_skills_activatable_via_tool(self): + """Test that sandbox-loaded skills can be activated via the skills tool.""" + sandbox = _make_mock_sandbox( + files={"/home/skills/my-skill/SKILL.md": SKILL_CONTENT}, + dirs={ + "/home/skills": [FileInfo(name="my-skill", is_dir=True)], + }, + ) + agent = _mock_agent(sandbox=sandbox) + + plugin = AgentSkills(skills=["sandbox:///home/skills"]) + await plugin.init_agent(agent) + + # Create tool context and activate the skill + tool_use = {"toolUseId": "test-id", "name": "skills", "input": {}} + tool_context = ToolContext(tool_use=tool_use, agent=agent, invocation_state={"agent": agent}) + + result = plugin.skills(skill_name="sandbox-skill", tool_context=tool_context) + + assert "Follow these steps" in result + + @pytest.mark.asyncio + async def test_no_sandbox_sources_stays_sync(self): + """Test that plugin with no sandbox sources works fine with sync init.""" + skill = Skill(name="local", description="Local skill", instructions="Do it") + plugin = AgentSkills(skills=[skill]) + agent = _mock_agent() + + # init_agent should work even though it's async + await plugin.init_agent(agent) + + assert "local" in plugin._get_skills(agent) + assert len(plugin._sandbox_sources) == 0 + + @pytest.mark.asyncio + async def test_sandbox_source_string_format(self): + """Test various sandbox: source string formats.""" + plugin = AgentSkills( + skills=[ + "sandbox:///home/skills", + "sandbox:///absolute/path/to/skills", + "sandbox:///tmp/skills/", + ] + ) + + assert len(plugin._sandbox_sources) == 3 + assert plugin._sandbox_sources[0] == "/home/skills" + assert plugin._sandbox_sources[1] == "/absolute/path/to/skills" + assert plugin._sandbox_sources[2] == "/tmp/skills/" + + +class TestAgentSkillsSandboxPluginRegistry: + """Test that sandbox skills work through the plugin registry (async init_agent).""" + + @pytest.mark.asyncio + async def test_plugin_registry_handles_async_init(self): + """Test that _PluginRegistry correctly handles async init_agent from AgentSkills.""" + from strands.plugins.registry import _PluginRegistry + + sandbox = _make_mock_sandbox( + files={"/home/skills/my-skill/SKILL.md": SKILL_CONTENT}, + dirs={ + "/home/skills": [FileInfo(name="my-skill", is_dir=True)], + }, + ) + agent = _mock_agent(sandbox=sandbox) + + plugin = AgentSkills(skills=["sandbox:///home/skills"]) + + # The registry should handle the async init_agent + registry = _PluginRegistry(agent) + registry.add_and_init(plugin) + + # After registry init, sandbox skills should be loaded for this agent + assert "sandbox-skill" in plugin._get_skills(agent) + + @pytest.mark.asyncio + async def test_get_available_skills_after_sandbox_load(self): + """Test get_available_skills returns sandbox-loaded skills.""" + sandbox = _make_mock_sandbox( + files={"/home/skills/my-skill/SKILL.md": SKILL_CONTENT}, + dirs={ + "/home/skills": [FileInfo(name="my-skill", is_dir=True)], + }, + ) + agent = _mock_agent(sandbox=sandbox) + + plugin = AgentSkills(skills=["sandbox:///home/skills"]) + await plugin.init_agent(agent) + + available = plugin.get_available_skills(agent) + assert len(available) == 1 + assert available[0].name == "sandbox-skill" + + +# --- Tests for multi-agent isolation --- + + +class TestMultiAgentIsolation: + """Tests proving that per-agent skill catalogs are isolated.""" + + @pytest.mark.asyncio + async def test_two_agents_different_sandbox_skills(self): + """Test that two agents with different sandboxes get different skill catalogs.""" + # Agent A has Docker sandbox with docker-skill + sandbox_a = _make_mock_sandbox( + files={"/home/skills/docker/SKILL.md": SKILL_C_CONTENT}, + dirs={"/home/skills": [FileInfo(name="docker", is_dir=True)]}, + ) + agent_a = _mock_agent(sandbox=sandbox_a) + + # Agent B has S3 sandbox with s3-skill + sandbox_b = _make_mock_sandbox( + files={"/home/skills/s3/SKILL.md": SKILL_D_CONTENT}, + dirs={"/home/skills": [FileInfo(name="s3", is_dir=True)]}, + ) + agent_b = _mock_agent(sandbox=sandbox_b) + + # SAME plugin instance + plugin = AgentSkills(skills=["sandbox:///home/skills"]) + + await plugin.init_agent(agent_a) + await plugin.init_agent(agent_b) + + # Agent A should ONLY see docker-skill + skills_a = plugin._get_skills(agent_a) + assert "docker-skill" in skills_a + assert "s3-skill" not in skills_a + + # Agent B should ONLY see s3-skill + skills_b = plugin._get_skills(agent_b) + assert "s3-skill" in skills_b + assert "docker-skill" not in skills_b + + @pytest.mark.asyncio + async def test_two_agents_share_base_skills(self): + """Test that base (sync) skills are shared across agents.""" + base_skill = Skill(name="shared-skill", description="Shared", instructions="Shared instructions") + + sandbox_a = _make_mock_sandbox( + files={"/skills/a/SKILL.md": SKILL_C_CONTENT}, + dirs={"/skills": [FileInfo(name="a", is_dir=True)]}, + ) + agent_a = _mock_agent(sandbox=sandbox_a) + + sandbox_b = _make_mock_sandbox( + files={"/skills/b/SKILL.md": SKILL_D_CONTENT}, + dirs={"/skills": [FileInfo(name="b", is_dir=True)]}, + ) + agent_b = _mock_agent(sandbox=sandbox_b) + + plugin = AgentSkills(skills=[base_skill, "sandbox:///skills"]) + + await plugin.init_agent(agent_a) + await plugin.init_agent(agent_b) + + # Both agents should have the shared base skill + assert "shared-skill" in plugin._get_skills(agent_a) + assert "shared-skill" in plugin._get_skills(agent_b) + + # But each should also have their own sandbox skill + assert "docker-skill" in plugin._get_skills(agent_a) + assert "docker-skill" not in plugin._get_skills(agent_b) + assert "s3-skill" in plugin._get_skills(agent_b) + assert "s3-skill" not in plugin._get_skills(agent_a) + + @pytest.mark.asyncio + async def test_two_agents_overlapping_skill_names_stay_isolated(self): + """Test that two agents with same skill name from different sandboxes get different content.""" + # Both sandboxes have a skill named "sandbox-skill" but with different content + sandbox_a = _make_mock_sandbox( + files={"/skills/a/SKILL.md": SKILL_CONTENT}, # sandbox-skill: "A skill loaded from sandbox" + dirs={"/skills": [FileInfo(name="a", is_dir=True)]}, + ) + agent_a = _mock_agent(sandbox=sandbox_a) + + modified_content = """--- +name: sandbox-skill +description: MODIFIED sandbox skill from agent B +--- +# Modified Instructions + +These are different instructions. +""" + sandbox_b = _make_mock_sandbox( + files={"/skills/b/SKILL.md": modified_content}, + dirs={"/skills": [FileInfo(name="b", is_dir=True)]}, + ) + agent_b = _mock_agent(sandbox=sandbox_b) + + plugin = AgentSkills(skills=["sandbox:///skills"]) + + await plugin.init_agent(agent_a) + await plugin.init_agent(agent_b) + + # Same name, different descriptions + assert plugin._get_skills(agent_a)["sandbox-skill"].description == "A skill loaded from sandbox" + assert plugin._get_skills(agent_b)["sandbox-skill"].description == "MODIFIED sandbox skill from agent B" + + @pytest.mark.asyncio + async def test_agent_skills_tool_uses_per_agent_catalog(self): + """Test that the skills tool returns the correct skill for each agent.""" + sandbox_a = _make_mock_sandbox( + files={"/skills/a/SKILL.md": SKILL_C_CONTENT}, + dirs={"/skills": [FileInfo(name="a", is_dir=True)]}, + ) + agent_a = _mock_agent(sandbox=sandbox_a) + + sandbox_b = _make_mock_sandbox( + files={"/skills/b/SKILL.md": SKILL_D_CONTENT}, + dirs={"/skills": [FileInfo(name="b", is_dir=True)]}, + ) + agent_b = _mock_agent(sandbox=sandbox_b) + + plugin = AgentSkills(skills=["sandbox:///skills"]) + await plugin.init_agent(agent_a) + await plugin.init_agent(agent_b) + + # Agent A can activate docker-skill but NOT s3-skill + tool_use = {"toolUseId": "test-id", "name": "skills", "input": {}} + ctx_a = ToolContext(tool_use=tool_use, agent=agent_a, invocation_state={"agent": agent_a}) + result_a = plugin.skills(skill_name="docker-skill", tool_context=ctx_a) + assert "Docker-specific instructions" in result_a + + result_a_missing = plugin.skills(skill_name="s3-skill", tool_context=ctx_a) + assert "not found" in result_a_missing + + # Agent B can activate s3-skill but NOT docker-skill + ctx_b = ToolContext(tool_use=tool_use, agent=agent_b, invocation_state={"agent": agent_b}) + result_b = plugin.skills(skill_name="s3-skill", tool_context=ctx_b) + assert "S3-specific instructions" in result_b + + result_b_missing = plugin.skills(skill_name="docker-skill", tool_context=ctx_b) + assert "not found" in result_b_missing + + @pytest.mark.asyncio + async def test_generate_skills_xml_per_agent(self): + """Test that _generate_skills_xml returns per-agent XML.""" + sandbox_a = _make_mock_sandbox( + files={"/skills/a/SKILL.md": SKILL_C_CONTENT}, + dirs={"/skills": [FileInfo(name="a", is_dir=True)]}, + ) + agent_a = _mock_agent(sandbox=sandbox_a) + + sandbox_b = _make_mock_sandbox( + files={"/skills/b/SKILL.md": SKILL_D_CONTENT}, + dirs={"/skills": [FileInfo(name="b", is_dir=True)]}, + ) + agent_b = _mock_agent(sandbox=sandbox_b) + + plugin = AgentSkills(skills=["sandbox:///skills"]) + await plugin.init_agent(agent_a) + await plugin.init_agent(agent_b) + + xml_a = plugin._generate_skills_xml(agent_a) + xml_b = plugin._generate_skills_xml(agent_b) + + assert "docker-skill" in xml_a + assert "s3-skill" not in xml_a + + assert "s3-skill" in xml_b + assert "docker-skill" not in xml_b + + @pytest.mark.asyncio + async def test_get_available_skills_with_agent(self): + """Test get_available_skills with agent parameter returns per-agent skills.""" + sandbox = _make_mock_sandbox( + files={"/skills/a/SKILL.md": SKILL_C_CONTENT}, + dirs={"/skills": [FileInfo(name="a", is_dir=True)]}, + ) + agent = _mock_agent(sandbox=sandbox) + + plugin = AgentSkills(skills=["sandbox:///skills"]) + await plugin.init_agent(agent) + + # With agent — returns per-agent catalog + with_agent = plugin.get_available_skills(agent) + assert len(with_agent) == 1 + assert with_agent[0].name == "docker-skill" + + # Without agent — returns base (empty since all sources are sandbox) + without_agent = plugin.get_available_skills() + assert len(without_agent) == 0 + + @pytest.mark.asyncio + async def test_base_skills_not_mutated_by_init_agent(self): + """Test that init_agent doesn't mutate base_skills.""" + base = Skill(name="base-skill", description="Base", instructions="Base") + sandbox = _make_mock_sandbox( + files={"/skills/a/SKILL.md": SKILL_C_CONTENT}, + dirs={"/skills": [FileInfo(name="a", is_dir=True)]}, + ) + agent = _mock_agent(sandbox=sandbox) + + plugin = AgentSkills(skills=[base, "sandbox:///skills"]) + + # Before init + assert list(plugin._base_skills.keys()) == ["base-skill"] + + await plugin.init_agent(agent) + + # Base skills should NOT have sandbox skills added + assert list(plugin._base_skills.keys()) == ["base-skill"] + # But per-agent should have both + assert "base-skill" in plugin._get_skills(agent) + assert "docker-skill" in plugin._get_skills(agent) + + @pytest.mark.asyncio + async def test_set_available_skills_clears_per_agent_caches(self): + """Test that set_available_skills clears per-agent caches.""" + sandbox = _make_mock_sandbox( + files={"/skills/a/SKILL.md": SKILL_C_CONTENT}, + dirs={"/skills": [FileInfo(name="a", is_dir=True)]}, + ) + agent = _mock_agent(sandbox=sandbox) + + plugin = AgentSkills(skills=["sandbox:///skills"]) + await plugin.init_agent(agent) + + assert "docker-skill" in plugin._get_skills(agent) + + # Replace skills — should clear per-agent cache + new_skill = Skill(name="new-skill", description="New", instructions="New") + plugin.set_available_skills([new_skill]) + + # Per-agent cache cleared — falls back to new base + assert "docker-skill" not in plugin._get_skills(agent) + assert "new-skill" in plugin._get_skills(agent) + + @pytest.mark.asyncio + async def test_no_sandbox_agent_gets_base_skills_only(self): + """Test that an agent without sandbox sources gets base skills only.""" + base = Skill(name="base-skill", description="Base", instructions="Base") + plugin = AgentSkills(skills=[base]) + agent = _mock_agent() + + await plugin.init_agent(agent) + + skills = plugin._get_skills(agent) + assert len(skills) == 1 + assert "base-skill" in skills