fix(swarm): enforce per-worker skills allowlist in load_skill

SwarmAgentSpec.skills documents a per-worker skill boundary, but the
boundary was only applied to the skill descriptions injected into the
worker prompt; the load_skill tool itself would load any skill by name,
so a worker could pull in documentation its preset never granted.

LoadSkillTool now accepts an optional allowed_skills set and fails
closed with the allowed names when a skill outside it is requested;
None keeps the historical unrestricted behavior for the main agent.
build_filtered_registry and build_swarm_registry take a matching
skill_allowlist parameter and rebuild load_skill with the boundary
baked in (register() replaces the auto-discovered instance by name),
and run_worker delivers agent_spec.skills through it. An empty skills
list remains unrestricted, matching the prompt-side filter semantics.

Also adapts the four build_swarm_registry test doubles in
test_swarm_m4_e2e.py to the new keyword-only parameter (this
adaptation was carried by the feature commit in #1286; the fix
is not self-contained against the current test suite without it).

Resubmitted standalone as requested in the #1286 review ("please do
re-submit it separately, and it does not need to wait on anything");
gap re-verified on main @ f9cb061b — models.py still documents the
boundary, the worker.py prompt-text filter is still the only reader
of spec.skills, and load_skill_tool.py still has no enforcement.

Signed-off-by: shadowinlife <shadowinlife@gmail.com>

agent/src/swarm/worker.py

@@ -573,11 +573,16 @@ def _run_worker_impl(
     _emit(event_callback, "worker_started", agent_id, task_id)
 
     # 1. Build per-worker tool registry — local pool plus any operator-
-    #    surfaced MCP tools, projected onto the agent's whitelist.
+    #    surfaced MCP tools, projected onto the agent's whitelist. The
+    #    documented ``skills:`` boundary is enforced at runtime by rebuilding
+    #    ``load_skill`` with the allowlist baked in; an empty ``skills`` list
+    #    means unrestricted, matching the prompt-side filter semantics
+    #    (``_filter_skill_descriptions`` treats empty as include-all).
     registry = build_swarm_registry(
         agent_spec.tools,
         agent_config=agent_config,
         include_shell_tools=include_shell_tools,
+        skill_allowlist=agent_spec.skills or None,
     )
 
     # 2. Build system prompt with filtered skills

agent/src/tools/__init__.py

@@ -275,7 +275,12 @@ def build_registry(
     return registry
 
 
-def build_filtered_registry(tool_names: list[str], *, include_shell_tools: bool = False) -> ToolRegistry:
+def build_filtered_registry(
+    tool_names: list[str],
+    *,
+    include_shell_tools: bool = False,
+    skill_allowlist: list[str] | None = None,
+) -> ToolRegistry:
     """Build a ToolRegistry with only specified tools.
 
     Local-tools-only filtered builder. Swarm workers should call
@@ -286,12 +291,20 @@ def build_filtered_registry(tool_names: list[str], *, include_shell_tools: bool
     Args:
         tool_names: Tool names to include.
         include_shell_tools: Whether to include filtered shell execution tools.
+        skill_allowlist: When not ``None``, the ``load_skill`` tool (if
+            whitelisted) is rebuilt to refuse skill names outside this list,
+            so a documented skill boundary is enforced at runtime.
 
     Returns:
         ToolRegistry containing only the requested tools.
     """
     full = build_registry(include_shell_tools=include_shell_tools)
-    return _filter_registry(full, tool_names, include_shell_tools=include_shell_tools)
+    return _filter_registry(
+        full,
+        tool_names,
+        include_shell_tools=include_shell_tools,
+        skill_allowlist=skill_allowlist,
+    )
 
 
 def build_swarm_registry(
@@ -299,6 +312,7 @@ def build_swarm_registry(
     *,
     agent_config: "AgentConfig | None" = None,
     include_shell_tools: bool = False,
+    skill_allowlist: list[str] | None = None,
 ) -> ToolRegistry:
     """Build a per-worker registry that merges local + remote MCP tools.
 
@@ -321,6 +335,9 @@ def build_swarm_registry(
             MCP wrappers are appended to the candidate pool before filtering.
             Pass ``None`` to keep the worker strictly local.
         include_shell_tools: Whether shell-execution tools are eligible.
+        skill_allowlist: When not ``None``, the ``load_skill`` tool (if
+            whitelisted) is rebuilt to refuse skill names outside this list,
+            enforcing the per-worker ``skills:`` boundary at runtime.
 
     Returns:
         ToolRegistry containing the whitelist intersection of local tools
@@ -335,7 +352,12 @@ def build_swarm_registry(
         include_shell_tools=include_shell_tools,
         _mcp_server_tool_name_segments=swarm_local_server_names,
     )
-    return _filter_registry(full, tool_names, include_shell_tools=include_shell_tools)
+    return _filter_registry(
+        full,
+        tool_names,
+        include_shell_tools=include_shell_tools,
+        skill_allowlist=skill_allowlist,
+    )
 
 
 def _prune_agent_config_for_swarm_tools(
@@ -374,6 +396,7 @@ def _filter_registry(
     tool_names: list[str],
     *,
     include_shell_tools: bool,
+    skill_allowlist: list[str] | None = None,
 ) -> ToolRegistry:
     """Project a full registry down to a whitelist with consistent drop logging."""
     filtered = ToolRegistry()
@@ -388,6 +411,12 @@ def _filter_registry(
                 "depends on it cannot execute it.",
                 name, include_shell_tools,
             )
+    if skill_allowlist is not None and filtered.get("load_skill") is not None:
+        # Rebuild load_skill with the boundary baked in. ``register`` replaces
+        # by name, so the unrestricted auto-discovered instance is dropped.
+        from src.tools.load_skill_tool import LoadSkillTool
+
+        filtered.register(LoadSkillTool(allowed_skills=frozenset(skill_allowlist)))
     return filtered
 
 

agent/src/tools/load_skill_tool.py

@@ -185,13 +185,24 @@ class LoadSkillTool(BaseTool):
     }
     repeatable = True
 
-    def __init__(self, skills_loader: SkillsLoader | None = None) -> None:
+    def __init__(
+        self,
+        skills_loader: SkillsLoader | None = None,
+        allowed_skills: "frozenset[str] | None" = None,
+    ) -> None:
         """Initialize LoadSkillTool.
 
         Args:
             skills_loader: SkillsLoader instance; creates one automatically if omitted.
+            allowed_skills: When not ``None``, only these skill names may be
+                loaded; any other name fails closed with an error listing the
+                allowed set. ``None`` (the default) keeps the historical
+                unrestricted behavior. Swarm workers receive their spec's
+                ``skills`` list here so the documented per-agent skill
+                boundary is enforced at runtime, not only in the prompt.
         """
         self._loader = skills_loader or SkillsLoader()
+        self._allowed_skills = allowed_skills
 
     def execute(self, **kwargs: Any) -> str:
         """Load a skill as a skeleton, a named section, or a character page.
@@ -211,6 +222,12 @@ class LoadSkillTool(BaseTool):
             KeyError: If ``name`` is missing.
         """
         name = kwargs["name"]
+        if self._allowed_skills is not None and name not in self._allowed_skills:
+            allowed = ", ".join(sorted(self._allowed_skills)) or "(none)"
+            return _error(
+                f"Error: skill {name!r} is outside the skill allowlist for this "
+                f"context. Allowed skills: {allowed}"
+            )
         content = self._loader.get_content(name)
         if content.startswith("Error:"):
             return json.dumps({"status": "error", "content": content}, ensure_ascii=False)

agent/tests/test_load_skill_allowlist.py

@@ -0,0 +1,217 @@
+"""Regression tests for the per-context ``load_skill`` skill allowlist.
+
+``SwarmAgentSpec.skills`` documents a per-worker skill boundary
+(``src/swarm/models.py``), but before this fix the boundary only filtered the
+skill *descriptions* shown in the worker prompt — the ``load_skill`` tool
+itself would load any skill by name, so the documented boundary was not
+enforced at runtime. ``LoadSkillTool(allowed_skills=...)`` closes that gap;
+these tests pin the three states (restricted / unrestricted / empty) plus the
+registry-builder and swarm-worker wiring that delivers the allowlist.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from unittest.mock import patch
+
+from src.agent.skills import SkillsLoader
+from src.swarm import worker as worker_mod
+from src.swarm.models import SwarmAgentSpec, SwarmTask
+from src.swarm.worker import run_worker
+from src.tools import build_filtered_registry, build_swarm_registry
+from src.tools.load_skill_tool import LoadSkillTool
+
+
+def _write_skill(root: Path, name: str) -> None:
+    skill_dir = root / name
+    skill_dir.mkdir(parents=True)
+    (skill_dir / "SKILL.md").write_text(
+        f"---\nname: {name}\ndescription: {name} description\n---\n\n# {name}\n\nBody.\n",
+        encoding="utf-8",
+    )
+
+
+def _fixture_loader(tmp_path: Path) -> SkillsLoader:
+    _write_skill(tmp_path, "alpha-one")
+    _write_skill(tmp_path, "beta-two")
+    return SkillsLoader(skills_dir=tmp_path, user_skills_dir=tmp_path / "absent")
+
+
+def test_allowlist_permits_listed_skill(tmp_path: Path) -> None:
+    tool = LoadSkillTool(
+        _fixture_loader(tmp_path), allowed_skills=frozenset({"alpha-one"})
+    )
+    payload = json.loads(tool.execute(name="alpha-one"))
+    assert payload["status"] == "ok"
+
+
+def test_allowlist_refuses_unlisted_skill_and_names_allowed_set(tmp_path: Path) -> None:
+    tool = LoadSkillTool(
+        _fixture_loader(tmp_path), allowed_skills=frozenset({"alpha-one"})
+    )
+    payload = json.loads(tool.execute(name="beta-two"))
+    assert payload["status"] == "error"
+    assert "outside the skill allowlist" in payload["content"]
+    assert "alpha-one" in payload["content"]
+
+
+def test_allowlist_none_keeps_unrestricted_default(tmp_path: Path) -> None:
+    tool = LoadSkillTool(_fixture_loader(tmp_path))
+    assert json.loads(tool.execute(name="alpha-one"))["status"] == "ok"
+    assert json.loads(tool.execute(name="beta-two"))["status"] == "ok"
+
+
+def test_allowlist_empty_refuses_everything(tmp_path: Path) -> None:
+    tool = LoadSkillTool(_fixture_loader(tmp_path), allowed_skills=frozenset())
+    payload = json.loads(tool.execute(name="alpha-one"))
+    assert payload["status"] == "error"
+    assert "(none)" in payload["content"]
+
+
+def test_filtered_registry_rebuilds_load_skill_with_allowlist() -> None:
+    registry = build_filtered_registry(
+        ["load_skill"], skill_allowlist=["strategy-generate"]
+    )
+    tool = registry.get("load_skill")
+    assert tool is not None
+    # An allowlisted bundled skill loads; any other bundled skill is refused.
+    assert json.loads(tool.execute(name="strategy-generate"))["status"] == "ok"
+    refused = json.loads(tool.execute(name="alpha-zoo"))
+    assert refused["status"] == "error"
+    assert "strategy-generate" in refused["content"]
+
+
+def test_filtered_registry_none_allowlist_keeps_unrestricted_instance() -> None:
+    registry = build_filtered_registry(["load_skill"])
+    tool = registry.get("load_skill")
+    assert tool is not None
+    assert json.loads(tool.execute(name="alpha-zoo"))["status"] == "ok"
+
+
+def test_allowlist_without_load_skill_in_whitelist_adds_nothing() -> None:
+    registry = build_filtered_registry(
+        ["read_file"], skill_allowlist=["strategy-generate"]
+    )
+    assert registry.get("load_skill") is None
+
+
+def test_swarm_registry_honours_skill_allowlist() -> None:
+    registry = build_swarm_registry(
+        ["load_skill"], skill_allowlist=["strategy-generate"]
+    )
+    tool = registry.get("load_skill")
+    assert tool is not None
+    refused = json.loads(tool.execute(name="alpha-zoo"))
+    assert refused["status"] == "error"
+    assert "strategy-generate" in refused["content"]
+
+
+def test_swarm_worker_passes_spec_skills_as_allowlist(tmp_path: Path) -> None:
+    """run_worker must deliver ``agent_spec.skills`` to the registry builder —
+    the one line that makes the documented per-worker boundary real."""
+    captured: dict = {}
+
+    class _CapturingRegistry:
+        def get_definitions(self) -> list[dict]:
+            return []
+
+        def get(self, name: str):
+            return None
+
+        def execute(self, name: str, args: dict) -> str:
+            return json.dumps({"status": "ok"})
+
+    class _FinalAnswerLLM:
+        def __call__(self, *args, **kwargs) -> "_FinalAnswerLLM":
+            return self
+
+        def close(self) -> None:
+            pass
+
+        def stream_chat(self, messages, tools=None, on_text_chunk=None, timeout=None):
+            from src.providers.llm import LLMResponse
+
+            return LLMResponse(content="done")
+
+    def _capture(*args, **kwargs):
+        captured.update(kwargs)
+        return _CapturingRegistry()
+
+    agent = SwarmAgentSpec(
+        id="analyst",
+        role="Analyst",
+        system_prompt="You analyze.",
+        tools=["load_skill"],
+        skills=["strategy-generate"],
+        max_iterations=1,
+        timeout_seconds=60,
+    )
+    task = SwarmTask(id="t1", agent_id="analyst", prompt_template="Do the thing.")
+    with (
+        patch.object(worker_mod, "build_swarm_registry", _capture),
+        patch.object(worker_mod, "ChatLLM", _FinalAnswerLLM()),
+    ):
+        run_worker(
+            agent_spec=agent,
+            task=task,
+            upstream_summaries={},
+            user_vars={},
+            run_dir=tmp_path,
+        )
+    assert captured.get("skill_allowlist") == ["strategy-generate"]
+
+
+def test_swarm_worker_empty_skills_stays_unrestricted(tmp_path: Path) -> None:
+    """An empty ``skills`` list means unrestricted (prompt-side filter treats
+    empty as include-all); the wiring must pass ``None``, not an empty list."""
+    captured: dict = {}
+
+    class _CapturingRegistry:
+        def get_definitions(self) -> list[dict]:
+            return []
+
+        def get(self, name: str):
+            return None
+
+        def execute(self, name: str, args: dict) -> str:
+            return json.dumps({"status": "ok"})
+
+    class _FinalAnswerLLM:
+        def __call__(self, *args, **kwargs) -> "_FinalAnswerLLM":
+            return self
+
+        def close(self) -> None:
+            pass
+
+        def stream_chat(self, messages, tools=None, on_text_chunk=None, timeout=None):
+            from src.providers.llm import LLMResponse
+
+            return LLMResponse(content="done")
+
+    def _capture(*args, **kwargs):
+        captured.update(kwargs)
+        return _CapturingRegistry()
+
+    agent = SwarmAgentSpec(
+        id="analyst",
+        role="Analyst",
+        system_prompt="You analyze.",
+        tools=["load_skill"],
+        skills=[],
+        max_iterations=1,
+        timeout_seconds=60,
+    )
+    task = SwarmTask(id="t1", agent_id="analyst", prompt_template="Do the thing.")
+    with (
+        patch.object(worker_mod, "build_swarm_registry", _capture),
+        patch.object(worker_mod, "ChatLLM", _FinalAnswerLLM()),
+    ):
+        run_worker(
+            agent_spec=agent,
+            task=task,
+            upstream_summaries={},
+            user_vars={},
+            run_dir=tmp_path,
+        )
+    assert captured.get("skill_allowlist") is None

agent/tests/test_swarm_m4_e2e.py

@@ -274,7 +274,7 @@ def test_run_worker_uses_remote_mcp_tool_and_report_cites_canned_data(
     with (
         patch(
             "src.swarm.worker.build_swarm_registry",
-            wraps=lambda tool_names, *, agent_config=None, include_shell_tools=False: _registry_with_remote(
+            wraps=lambda tool_names, *, agent_config=None, include_shell_tools=False, skill_allowlist=None: _registry_with_remote(
                 tool_names,
                 remote_tools,
                 include_shell_tools=include_shell_tools,
@@ -358,7 +358,7 @@ def test_two_agents_with_distinct_servers_only_see_their_own_remote_tools(
         with (
             patch(
                 "src.swarm.worker.build_swarm_registry",
-                wraps=lambda tool_names, *, agent_config=None, include_shell_tools=False: _registry_with_remote(
+                wraps=lambda tool_names, *, agent_config=None, include_shell_tools=False, skill_allowlist=None: _registry_with_remote(
                     tool_names, remote_tools, include_shell_tools=include_shell_tools,
                 ),
             ),
@@ -453,7 +453,7 @@ def test_tool_call_events_carry_mcp_metadata_and_redact_sensitive_arguments(
     with (
         patch(
             "src.swarm.worker.build_swarm_registry",
-            wraps=lambda tool_names, *, agent_config=None, include_shell_tools=False: _registry_with_remote(
+            wraps=lambda tool_names, *, agent_config=None, include_shell_tools=False, skill_allowlist=None: _registry_with_remote(
                 tool_names, remote_tools, include_shell_tools=include_shell_tools,
             ),
         ),
@@ -543,7 +543,7 @@ def test_remote_tool_transport_failure_does_not_crash_worker(tmp_path: Path) ->
     with (
         patch(
             "src.swarm.worker.build_swarm_registry",
-            wraps=lambda tool_names, *, agent_config=None, include_shell_tools=False: _registry_with_remote(
+            wraps=lambda tool_names, *, agent_config=None, include_shell_tools=False, skill_allowlist=None: _registry_with_remote(
                 tool_names, remote_tools, include_shell_tools=include_shell_tools,
             ),
         ),