feat(agents,claude_sdk): skill sets per conversation kind

src/beaver_gateway/agents/claude.py

@@ -3,8 +3,10 @@
 The system prompt is ``system_prompt`` verbatim or, per conversation kind,
 the granules named in ``prompts`` assembled at every session spawn (see
 ``core/prompt.py``). ``skill_sets`` are directories of ``<skill>/SKILL.md``
-folders; each becomes a local SDK plugin. Nothing from disk is loaded
-otherwise: the adapter runs with ``setting_sources=[]``.
+folders; each becomes a local SDK plugin, either the same tuple for every
+kind or a ``SkillSets`` with a tuple per kind (§4.3: the master never sees
+what a branch opens). Nothing from disk is loaded otherwise: the adapter
+runs with ``setting_sources=[]``.
 """
 
 from __future__ import annotations
@@ -19,7 +21,7 @@ from beaver_gateway.core.kinds import KINDS, Kind
 from beaver_gateway.core.policy import PolicyRule  # noqa: TC001 - pydantic runtime
 from beaver_gateway.core.prompt import PromptSource  # noqa: TC001 - pydantic runtime
 
-__all__ = ["ClaudeAgent", "ClaudeOptions", "Prompts"]
+__all__ = ["ClaudeAgent", "ClaudeOptions", "Prompts", "SkillSets"]
 
 
 class ClaudeOptions(BaseModel):
@@ -70,6 +72,21 @@ class Prompts(BaseModel):
         return tuple(k for k in KINDS if self.for_kind(k) is not None)
 
 
+class SkillSets(BaseModel):
+    """Skill-set directories per conversation kind; ``None`` means none."""
+
+    model_config = ConfigDict(frozen=True)
+
+    master: tuple[Path, ...] | None = None
+    branch: tuple[Path, ...] | None = None
+    deep: tuple[Path, ...] | None = None
+    job: tuple[Path, ...] | None = None
+    fork: tuple[Path, ...] | None = None
+
+    def for_kind(self, kind: Kind) -> tuple[Path, ...]:
+        return getattr(self, kind) or ()
+
+
 class ClaudeAgent(BaseAgent):
     cwd: Path
     system_prompt: str = ""
@@ -81,7 +98,11 @@ class ClaudeAgent(BaseAgent):
     rest. Defaults to the kinds ``prompts`` covers, or ``("deep",)`` for a
     verbatim ``system_prompt``."""
 
-    skill_sets: tuple[Path, ...] = ()
+    skill_sets: tuple[Path, ...] | SkillSets = ()
+    """Skill-set directories, each a local plugin: one tuple for every kind,
+    or ``SkillSets`` to give each kind its own (a kind left ``None`` gets
+    no skills)."""
+
     gateway_tools: tuple[str, ...] = ()
     """Gateway tools exposed in-process (``read_conversation``, ``spawn``,
     ``say``, ``schedule``, ``inject``); empty = no gateway MCP server."""
@@ -106,5 +127,10 @@ class ClaudeAgent(BaseAgent):
     def prompt_for(self, kind: Kind) -> tuple[PromptSource, ...] | None:
         return self.prompts.for_kind(kind)
 
+    def skills_for(self, kind: Kind) -> tuple[Path, ...]:
+        if isinstance(self.skill_sets, SkillSets):
+            return self.skill_sets.for_kind(kind)
+        return self.skill_sets
+
     def serves(self, kind: str) -> bool:
         return kind in self.kinds

src/beaver_gateway/backends/claude_sdk.py

@@ -108,6 +108,7 @@ if TYPE_CHECKING:
     from beaver_gateway.agents.base import BaseAgent
     from beaver_gateway.agents.claude import ClaudeAgent
     from beaver_gateway.core.events import MessageStreamEvent
+    from beaver_gateway.core.kinds import Kind
 
 
 _log = logging.getLogger("beaver_gateway.backends.claude_sdk")
@@ -574,8 +575,9 @@ class ClaudeSdkBackend:
         if self._runner.home is not None:
             env["HOME"] = str(self._runner.home)
             env.setdefault("CLAUDE_CONFIG_DIR", str(self._runner.home / ".claude"))
-        plugins = self._plugins()
-        sources = agent.prompt_for(as_kind(spec.kind))
+        kind = as_kind(spec.kind)
+        plugins = self._plugins(kind)
+        sources = agent.prompt_for(kind)
         system_prompt = (
             prompt_assembly.assemble(sources) if sources else agent.system_prompt
         )
@@ -683,10 +685,10 @@ class ClaudeSdkBackend:
 
         return {"PreToolUse": [HookMatcher(hooks=[cast("Any", pre_tool_use)])]}
 
-    def _plugins(self) -> list[dict[str, str]]:
+    def _plugins(self, kind: Kind) -> list[dict[str, str]]:
         plugins: list[dict[str, str]] = []
-        root = self._work_dir / "plugins" / self._agent.name
-        for raw in sorted(self._agent.skill_sets, key=str):
+        root = self._work_dir / "plugins" / self._agent.name / kind
+        for raw in sorted(self._agent.skills_for(kind), key=str):
             source = Path(str(raw))
             name = source.name
             target = root / name

tests/test_claude_sdk_backend.py

@@ -25,7 +25,7 @@ from claude_agent_sdk import (
 )
 
 from beaver_gateway.agents.base import ExposedMcp
-from beaver_gateway.agents.claude import ClaudeAgent, ClaudeOptions, Prompts
+from beaver_gateway.agents.claude import ClaudeAgent, ClaudeOptions, Prompts, SkillSets
 from beaver_gateway.backends.claude_sdk import (
     ClaudeSdkBackend,
     RunnerConfig,
@@ -355,29 +355,58 @@ async def test_mcp_deny_and_usage_sink(cwd: Path) -> None:
     assert seen[0].session_id == "fresh-session"
 
 
-async def test_skill_sets_become_sorted_plugins(cwd: Path) -> None:
+def _skill_sets(cwd: Path, *names: str) -> tuple[Path, ...]:
     sets = cwd / "skills"
-    for name in ("zeta", "общие"):
+    for name in names:
         (sets / name / "demo").mkdir(parents=True)
         (sets / name / "demo" / "SKILL.md").write_text("---\nname: demo\n---\n")
-    backend = _backend(
-        cwd, InMemorySessionStore(), skill_sets=(sets / "zeta", sets / "общие")
-    )
+    return tuple(sets / name for name in names)
+
+
+async def _plugins_for(backend: ClaudeSdkBackend, kind: str) -> list[Path]:
     await _drain(
         backend.complete(
             agent=backend.agent,
             messages=[{"role": "user", "content": "x"}],
-            conversation_id="c",
+            conversation_id=f"c-{kind}",
+            kind=kind,
         )
     )
-    plugins = FakeClient.instances[0].options.plugins
-    assert [p["type"] for p in plugins] == ["local", "local"]
+    plugins = FakeClient.instances[-1].options.plugins
+    assert all(p["type"] == "local" for p in plugins)
     paths = [Path(p["path"]) for p in plugins]
-    assert [p.name for p in paths] == ["zeta", "общие"]
     for path in paths:
+        assert path.parent.name == kind
+        assert path.parent.parent.name == backend.agent.name
         assert (path / ".claude-plugin" / "plugin.json").exists()
         assert (path / "skills" / "demo" / "SKILL.md").exists()
+    return paths
+
+
+async def test_skill_sets_become_sorted_plugins(cwd: Path) -> None:
+    zeta, common = _skill_sets(cwd, "zeta", "общие")
+    backend = _backend(cwd, InMemorySessionStore(), skill_sets=(zeta, common))
+    paths = await _plugins_for(backend, "deep")
+    assert [p.name for p in paths] == ["zeta", "общие"]
     assert FakeClient.instances[0].options.skills == "all"
+    assert [p.name for p in await _plugins_for(backend, "job")] == ["zeta", "общие"]
+
+
+async def test_skill_sets_per_kind(cwd: Path) -> None:
+    a, b = _skill_sets(cwd, "a", "b")
+    backend = _backend(
+        cwd,
+        InMemorySessionStore(),
+        kinds=("master", "branch", "fork"),
+        skill_sets=SkillSets(master=(a,), branch=(a, b)),
+    )
+    master = await _plugins_for(backend, "master")
+    branch = await _plugins_for(backend, "branch")
+    assert [p.name for p in master] == ["a"]
+    assert [p.name for p in branch] == ["a", "b"]
+    assert master[0] != branch[0]
+    assert await _plugins_for(backend, "fork") == []
+    assert FakeClient.instances[-1].options.skills is None
 
 
 async def test_prompt_sources_are_assembled(cwd: Path) -> None: