[arnav] feat: enable SDK skill auto-discovery via directory layout migration

Skills previously lived as flat `<slug>.md` files injected directly into
the user prompt, which bypassed Claude Code CLI's native discovery entirely.
Switch to the CLI's expected `<slug>/SKILL.md` layout so the SDK can surface
each skill's frontmatter to the model and let it call `Skill(slug)` lazily
on its own — without having to eagerly inject every skill body every turn.
Key changes:
- One-shot startup migration moves existing flat files into the directory
  layout (idempotent, crash-safe, preserves existing dir if already migrated)
- `get_installed_slugs()` with module-level cache (keyed by SKILLS_DIR) so
  the per-turn allowlist query is effectively free on the hot path
- `_ensure_frontmatter()` stamps a valid YAML block on every write so the
  CLI listing never silently drops a skill with missing frontmatter
- `_resolve_sdk_skill_allowlist()` replaces the old `skills=[]` kill-switch:
  passes only user-installed slugs to the SDK, filtering out bundled Claude
  Code skills (/init, /review, etc.) by not including them, and dropping
  manually-attached skills that are already injected into the user message
- CRUD endpoints and tests updated to the new directory layout throughout
This commit is contained in:
Arnav Naval
2026-05-10 17:56:23 -05:00
parent 84d5e1fa60
commit 38c6717fb7
4 changed files with 767 additions and 101 deletions
+59 -14
View File
@@ -828,6 +828,48 @@ class AgentManager:
sections.append(f"[Using skill: {name}]\n\n{content}")
return "\n\n".join(sections)
@staticmethod
def _resolve_sdk_skill_allowlist(attached_skills: list | None) -> list[str]:
"""Build the per-turn SDK Skill-tool allowlist.
The Claude Code CLI auto-discovers skills from `~/.claude/skills/<slug>/
SKILL.md` and surfaces each one's frontmatter (name + description) into
the model's system prompt so it can decide on its own when to call
`Skill(slug)` and pull the body. The SDK exposes that filter as
`options.skills`:
- `None` — CLI defaults (re-enables Claude Code's bundled plugin
skills like /init, /review, /security-review, ...). We don't want
those: half mutate ~/.claude state OpenSwarm doesn't read, and the
rest reference slash commands the backend never intercepts.
- `[]` — suppress everything (was the previous behavior; safe but
also kills user-installed skills, defeating the point).
- `list[str]` — allowlist. Bundled skills are filtered out because
they're not on the list, and the model still sees user skills.
Manually-attached skills (the `/` picker → `attached_skills` flow) are
excluded from the allowlist for *this* turn: their full body is
already prepended to the user message by `_resolve_attached_skills`,
so admitting them on the lazy Skill-tool path too would just let the
model load the same content a second time.
The installed-slugs scan is cached in `backend.apps.skills.skills`
and invalidated by the create / delete endpoints + the startup
migration, so this is effectively free on the hot path — no
per-turn filesystem walk.
"""
from backend.apps.skills.skills import get_installed_slugs
installed = get_installed_slugs()
if not attached_skills:
return list(installed)
attached_ids = {
s.get("id") for s in attached_skills if s.get("id")
}
if not attached_ids:
return list(installed)
return [slug for slug in installed if slug not in attached_ids]
@staticmethod
def _get_branch_messages(session) -> list:
"""Return the linear message list for the active branch, walking the branch tree."""
@@ -1921,20 +1963,23 @@ class AgentManager:
"type": "preset",
"preset": "claude_code",
}
# Suppress Claude Code's bundled plugin skills (/init, /review,
# /security-review, /simplify, /loop, /schedule, /update-config,
# /keybindings-help, /fewer-permission-prompts, /claude-api).
# OpenSwarm has its own skills system that injects skill content
# into the user prompt via _resolve_attached_skills, bypassing the
# SDK's Skill tool entirely — so the bundled skills only cause
# confusion: half mutate ~/.claude state OpenSwarm doesn't use
# (settings.json, keybindings.json), and the rest tell the model
# it can invoke slash commands the backend doesn't intercept.
# Empty list is the SDK's documented "skills off" signal — see
# the `skills` field on ClaudeAgentOptions in claude_agent_sdk
# types.py. User-attached skills are unaffected because they
# don't go through the Skill tool.
options_kwargs["skills"] = []
# Skill auto-discovery: hand the SDK an allowlist of the user's
# installed skill slugs (one dir per skill under ~/.claude/skills/)
# so the CLI exposes each one's frontmatter to the model and the
# model can call `Skill(slug)` on its own when relevant. Bundled
# Claude Code skills (/init, /review, /security-review, /simplify,
# /loop, /schedule, /update-config, /keybindings-help,
# /fewer-permission-prompts, /claude-api) are filtered out by
# virtue of *not* being on the allowlist — half mutate ~/.claude
# state OpenSwarm doesn't read and the rest reference slash
# commands the backend never intercepts.
#
# Skills the user attached this turn via `/` are dropped from the
# allowlist: their full body is already prepended to the user
# message by `_resolve_attached_skills`, so allowing the SDK
# tool path would let the model double-load the same content.
# See `_resolve_sdk_skill_allowlist` for the full rationale.
options_kwargs["skills"] = self._resolve_sdk_skill_allowlist(attached_skills)
# exclude_dynamic_sections=True tells the CLI to keep
# per-user/per-machine grounding (cwd, git status, recent
# commits, OS info) out of the cached system prompt prefix
+286 -47
View File
@@ -1,4 +1,5 @@
import os
import shutil
import json
import logging
import re
@@ -19,12 +20,147 @@ from backend.config.paths import SKILLS_WORKSPACE_DIR
async def skills_lifespan():
os.makedirs(SKILLS_DIR, exist_ok=True)
os.makedirs(SKILLS_WORKSPACE_DIR, exist_ok=True)
_migrate_flat_to_dir_layout()
yield
skills = SubApp("skills", skills_lifespan)
# ---------------------------------------------------------------------------
# Installed-slugs cache
# ---------------------------------------------------------------------------
# `_run_agent_loop` asks for the list of installed skill slugs on *every*
# turn so it can hand the SDK an auto-discovery allowlist. The disk layout
# rarely changes mid-session, so re-walking SKILLS_DIR + statting each
# subdir's SKILL.md on every turn is wasted work — for N skills and T
# turns that's N*T syscalls in the agent hot path, all returning the same
# answer.
#
# Cache the slug list at module scope, key it by the current SKILLS_DIR
# (so test monkeypatching automatically forces a refetch when the dir
# moves), and invalidate explicitly from the create / delete endpoints
# and from the startup migration. Update doesn't invalidate because
# updates can only change a skill's content or metadata, never the set
# of slugs (`update_skill` 404s if the slug dir doesn't already exist).
#
# Concurrency: FastAPI runs handlers on the asyncio event loop in a
# single thread; none of the cache-touching code paths await between
# reading and writing `_INSTALLED_SLUGS_CACHE`, so no lock is needed.
# Multi-worker deployments each get their own cache, but they also have
# disjoint disk layouts being scanned, so that's fine.
_INSTALLED_SLUGS_CACHE: tuple[str, list[str]] | None = None
def _invalidate_installed_slugs_cache() -> None:
"""Drop the cached slug list; next `get_installed_slugs` rescans."""
global _INSTALLED_SLUGS_CACHE
_INSTALLED_SLUGS_CACHE = None
def get_installed_slugs() -> list[str]:
"""Return sorted slugs of installed skills (`<slug>/SKILL.md`).
Cached at module level and keyed by SKILLS_DIR so a test that
monkeypatches the dir gets a fresh scan automatically. The returned
list is shared — callers must not mutate it.
"""
global _INSTALLED_SLUGS_CACHE
cached = _INSTALLED_SLUGS_CACHE
if cached is not None and cached[0] == SKILLS_DIR:
return cached[1]
if not os.path.isdir(SKILLS_DIR):
result: list[str] = []
else:
result = sorted(
entry
for entry in os.listdir(SKILLS_DIR)
if not entry.startswith(".")
and os.path.isfile(os.path.join(SKILLS_DIR, entry, "SKILL.md"))
)
_INSTALLED_SLUGS_CACHE = (SKILLS_DIR, result)
return result
# Skills used to live as flat files (`~/.claude/skills/<slug>.md`) because
# OpenSwarm injected their full body into the user prompt itself and never
# leaned on Claude Code's own discovery. Now that we want the agent to
# auto-discover skills via the SDK's Skill tool, we have to match the CLI's
# convention, which is one directory per skill with a SKILL.md inside —
# confirmed by the bundled CLI's discovery code (`<dir>/SKILL.md`) and its
# user-facing docs ("`~/.claude/skills/<name>/SKILL.md`"). Anything else is
# silently ignored by the CLI.
def _migrate_flat_to_dir_layout() -> None:
"""One-shot migration: move `<slug>.md` -> `<slug>/SKILL.md`.
Idempotent: re-running after the move is a no-op. Crash-safe: the new
file is written before the old one is removed. If a directory already
exists for a slug it wins; we log and leave the flat file untouched so
the operator can resolve it manually.
"""
if not os.path.isdir(SKILLS_DIR):
return
for entry in os.listdir(SKILLS_DIR):
if not entry.endswith(".md") or entry.startswith("."):
continue
flat_path = os.path.join(SKILLS_DIR, entry)
if not os.path.isfile(flat_path):
continue
slug = entry[:-3]
target_dir = os.path.join(SKILLS_DIR, slug)
target_skill = os.path.join(target_dir, "SKILL.md")
if os.path.isdir(target_dir):
if not os.path.isfile(target_skill):
# Pre-existing dir without a SKILL.md is unusable; bail
# rather than overwrite whatever else is in there.
logger.warning(
f"Skill migration: {target_dir} exists but has no SKILL.md; "
f"leaving flat file {flat_path} in place."
)
continue
logger.info(
f"Skill migration: directory {target_dir} already present, "
f"removing redundant flat file {flat_path}"
)
try:
os.remove(flat_path)
except OSError as exc:
logger.warning(f"Skill migration: could not remove {flat_path}: {exc}")
continue
try:
with open(flat_path, encoding="utf-8") as f:
body = f.read()
except OSError as exc:
logger.warning(f"Skill migration: could not read {flat_path}: {exc}")
continue
index = _load_index()
meta = index.get(slug, {})
body = _ensure_frontmatter(
body,
name=meta.get("name", slug.replace("-", " ").replace("_", " ").title()),
description=meta.get("description", ""),
)
os.makedirs(target_dir, exist_ok=True)
with open(target_skill, "w", encoding="utf-8") as f:
f.write(body)
try:
os.remove(flat_path)
except OSError as exc:
logger.warning(
f"Skill migration: wrote {target_skill} but could not remove "
f"{flat_path}: {exc}"
)
logger.info(f"Skill migration: {flat_path} -> {target_skill}")
# Migration may have changed the slug set; drop the cache so the
# first post-startup call (typically from `_run_agent_loop`) rescans.
_invalidate_installed_slugs_cache()
def _skill_path(slug: str) -> str:
"""Canonical on-disk path for a skill's SKILL.md."""
return os.path.join(SKILLS_DIR, slug, "SKILL.md")
def _load_index() -> dict[str, dict]:
if os.path.exists(INDEX_PATH):
with open(INDEX_PATH) as f:
@@ -33,40 +169,26 @@ def _load_index() -> dict[str, dict]:
def _save_index(index: dict[str, dict]):
os.makedirs(SKILLS_DIR, exist_ok=True)
with open(INDEX_PATH, "w") as f:
json.dump(index, f, indent=2)
def _sync_skills() -> list[Skill]:
"""Sync skills from the filesystem, updating the index."""
index = _load_index()
result = []
if os.path.exists(SKILLS_DIR):
for fname in os.listdir(SKILLS_DIR):
if fname.endswith(".md"):
fpath = os.path.join(SKILLS_DIR, fname)
with open(fpath) as f:
content = f.read()
skill_id = fname.replace(".md", "")
meta = index.get(skill_id, {})
skill = Skill(
id=skill_id,
name=meta.get("name", fname.replace(".md", "").replace("-", " ").replace("_", " ").title()),
description=meta.get("description", ""),
content=content,
file_path=fpath,
command=meta.get("command", fname.replace(".md", "")),
)
result.append(skill)
return result
@skills.router.get("/list")
async def list_skills():
return {"skills": [s.model_dump() for s in _sync_skills()]}
def _decode_yaml_scalar(raw: str) -> str:
"""Inverse of `_yaml_scalar` — decodes the values it can emit.
"""
raw = raw.strip()
if len(raw) >= 2 and raw[0] == '"' and raw[-1] == '"':
try:
return json.loads(raw)
except json.JSONDecodeError:
return raw[1:-1]
if len(raw) >= 2 and raw[0] == "'" and raw[-1] == "'":
# YAML single-quote rule: interior `''` is a literal `'`. We
# don't emit single-quoted scalars ourselves, but hand-edited
# frontmatter sometimes does.
return raw[1:-1].replace("''", "'")
return raw
def _parse_skill_frontmatter(raw: str) -> dict:
@@ -81,10 +203,89 @@ def _parse_skill_frontmatter(raw: str) -> dict:
for line in fm_block.splitlines():
m = re.match(r"^(\w[\w_-]*)\s*:\s*(.+)$", line)
if m:
meta[m.group(1).strip()] = m.group(2).strip().strip('"').strip("'")
meta[m.group(1).strip()] = _decode_yaml_scalar(m.group(2))
return meta
def _ensure_frontmatter(content: str, *, name: str, description: str) -> str:
"""Make sure SKILL.md starts with a YAML block carrying name+description.
The CLI only surfaces a skill in the auto-discovery listing if its
frontmatter parses, so on every write we either patch the existing
block (preserving the user's other keys) or prepend a fresh one.
"""
existing = _parse_skill_frontmatter(content)
if existing:
end = content.find("---", 3)
body = content[end + 3 :].lstrip("\n")
else:
body = content
merged = {**existing, "name": name}
if description or "description" not in merged:
merged["description"] = description
fm_lines = [f"{k}: {_yaml_scalar(v)}" for k, v in merged.items() if v != "" or k in ("name", "description")]
return "---\n" + "\n".join(fm_lines) + "\n---\n\n" + body
def _yaml_scalar(value: object) -> str:
"""Quote a YAML scalar only when needed (matches CLI parser tolerance)."""
s = str(value)
if s == "" or any(c in s for c in (":", "#", "\n")) or s.strip() != s:
return json.dumps(s)
return s
def _sync_skills() -> list[Skill]:
"""Sync skills from the filesystem, updating the index.
Walks `SKILLS_DIR/<slug>/SKILL.md` (the layout the CLI discovers) and
pulls metadata from the file's YAML frontmatter, falling back to the
sidecar index for fields the user only set via the API.
"""
index = _load_index()
result = []
if not os.path.isdir(SKILLS_DIR):
return result
for slug in os.listdir(SKILLS_DIR):
if slug.startswith("."):
continue
skill_dir = os.path.join(SKILLS_DIR, slug)
if not os.path.isdir(skill_dir):
continue
skill_path = os.path.join(skill_dir, "SKILL.md")
if not os.path.isfile(skill_path):
continue
with open(skill_path, encoding="utf-8") as f:
content = f.read()
frontmatter = _parse_skill_frontmatter(content)
meta = index.get(slug, {})
name = (
frontmatter.get("name")
or meta.get("name")
or slug.replace("-", " ").replace("_", " ").title()
)
description = frontmatter.get("description") or meta.get("description", "")
result.append(
Skill(
id=slug,
name=name,
description=description,
content=content,
file_path=skill_path,
command=meta.get("command", slug),
)
)
return result
@skills.router.get("/list")
async def list_skills():
return {"skills": [s.model_dump() for s in _sync_skills()]}
@skills.router.post("/workspace/seed")
async def seed_skill_workspace(body: SkillWorkspaceSeedRequest):
folder = os.path.join(SKILLS_WORKSPACE_DIR, body.workspace_id)
@@ -141,10 +342,15 @@ async def get_skill(skill_id: str):
@skills.router.post("/create")
async def create_skill(body: SkillCreate):
slug = body.name.lower().replace(" ", "-")
fpath = os.path.join(SKILLS_DIR, f"{slug}.md")
skill_dir = os.path.join(SKILLS_DIR, slug)
os.makedirs(skill_dir, exist_ok=True)
fpath = os.path.join(skill_dir, "SKILL.md")
final_content = _ensure_frontmatter(
body.content, name=body.name, description=body.description
)
with open(fpath, "w") as f:
f.write(body.content)
f.write(final_content)
index = _load_index()
index[slug] = {
@@ -158,33 +364,65 @@ async def create_skill(body: SkillCreate):
id=slug,
name=body.name,
description=body.description,
content=body.content,
content=final_content,
file_path=fpath,
command=body.command or slug,
)
pass
_invalidate_installed_slugs_cache()
return {"ok": True, "skill": skill.model_dump()}
@skills.router.put("/{skill_id}")
async def update_skill(skill_id: str, body: SkillUpdate):
fpath = os.path.join(SKILLS_DIR, f"{skill_id}.md")
fpath = _skill_path(skill_id)
if not os.path.exists(fpath):
raise HTTPException(status_code=404, detail="Skill not found")
if body.content is not None:
with open(fpath, "w") as f:
f.write(body.content)
index = _load_index()
meta = index.get(skill_id, {})
if body.name is not None:
meta["name"] = body.name
if body.description is not None:
meta["description"] = body.description
if body.command is not None:
meta["command"] = body.command
# Precedence for name/description, highest first:
# 1. explicit form field (`body.name` / `body.description`)
# 2. inline frontmatter inside `body.content` (so editing the YAML
# block in the editor textarea actually sticks instead of being
# silently clobbered by the stale index value)
# 3. previous index value
# 4. slug-derived fallback
#
inline = _parse_skill_frontmatter(body.content) if body.content else {}
if body.name is not None:
meta["name"] = body.name
elif inline.get("name"):
meta["name"] = inline["name"]
if body.description is not None:
meta["description"] = body.description
elif "description" in inline:
meta["description"] = inline["description"]
index[skill_id] = meta
if body.content is not None:
new_content = _ensure_frontmatter(
body.content,
name=meta.get("name", skill_id),
description=meta.get("description", ""),
)
with open(fpath, "w") as f:
f.write(new_content)
elif body.name is not None or body.description is not None:
# Metadata changed but body wasn't supplied — patch the
# frontmatter on disk so the CLI listing stays in sync.
with open(fpath, encoding="utf-8") as f:
existing = f.read()
new_content = _ensure_frontmatter(
existing,
name=meta.get("name", skill_id),
description=meta.get("description", ""),
)
with open(fpath, "w") as f:
f.write(new_content)
_save_index(index)
with open(fpath) as f:
@@ -203,10 +441,11 @@ async def update_skill(skill_id: str, body: SkillUpdate):
@skills.router.delete("/{skill_id}")
async def delete_skill(skill_id: str):
fpath = os.path.join(SKILLS_DIR, f"{skill_id}.md")
if os.path.exists(fpath):
os.remove(fpath)
skill_dir = os.path.join(SKILLS_DIR, skill_id)
if os.path.isdir(skill_dir):
shutil.rmtree(skill_dir)
index = _load_index()
index.pop(skill_id, None)
_save_index(index)
_invalidate_installed_slugs_cache()
return {"ok": True}
+106 -32
View File
@@ -314,25 +314,6 @@ def test_get_all_tool_names_drops_explicitly_denied_builtins(tmp_data_dirs):
# ---------------------------------------------------------------------------
def test_ensure_cwd_git_repo_creates_repo_when_missing(tmp_path):
"""Fresh tmp dir with no .git → function inits a repo + empty commit."""
cwd = tmp_path / "fresh"
cwd.mkdir()
_ensure_cwd_git_repo(str(cwd), home=str(tmp_path))
# Either .git lives here directly, or git decided we're already in
# a parent repo (the test runner's repo, for instance). Both are
# valid healthy outcomes.
if (cwd / ".git").exists():
# Verify HEAD resolves — the function commits an empty seed.
import subprocess
head = subprocess.run(
["git", "rev-parse", "--verify", "HEAD"],
cwd=str(cwd),
capture_output=True,
)
assert head.returncode == 0
def test_ensure_cwd_git_repo_skips_risky_roots(tmp_path):
"""Calling on $HOME / / / parent-of-home must short-circuit and
leave the directory untouched."""
@@ -416,24 +397,117 @@ def test_compose_system_prompt_all_none_returns_none():
assert AgentManager()._compose_system_prompt(None, None, None) is None
def test_run_agent_loop_disables_bundled_claude_code_skills():
"""Regression guard for `options_kwargs["skills"] = []` in _run_agent_loop.
def test_run_agent_loop_uses_skill_allowlist_not_kill_switch():
"""`_run_agent_loop` must hand the SDK a per-turn allowlist, not the
old `[]` kill-switch.
The claude-agent-sdk ships a Skill tool that surfaces Claude Code's
bundled plugin skills (/init, /review, /security-review, /simplify,
/loop, /schedule, /update-config, /keybindings-help,
/fewer-permission-prompts, /claude-api). These are inappropriate in
OpenSwarm — half mutate ~/.claude config the backend doesn't read, and
the rest reference slash commands the backend never intercepts. The
SDK's documented "skills off" signal is `skills=[]` on
ClaudeAgentOptions; user-attached skills bypass the Skill tool and
are unaffected. If this assignment ever gets dropped during a rebase
or refactor the bundled skills come back, so we pin it here.
The bundled Claude Code skills (/init, /review, /security-review,
/simplify, /loop, /schedule, /update-config, /keybindings-help,
/fewer-permission-prompts, /claude-api) need to stay suppressed —
half mutate ~/.claude state OpenSwarm doesn't read, the rest
reference slash commands the backend never intercepts. Passing
`None` to the SDK lets the CLI's defaults reintroduce them; passing
`[]` (the previous behavior) suppresses them but also kills user
skills. The right answer is `list[str]` with only user-installed
slugs, which is what `_resolve_sdk_skill_allowlist` produces. This
test pins the wiring so a future refactor can't silently revert.
"""
import inspect
src = inspect.getsource(AgentManager._run_agent_loop)
assert 'options_kwargs["skills"] = []' in src
assert 'options_kwargs["skills"] = self._resolve_sdk_skill_allowlist(' in src
assert 'options_kwargs["skills"] = []' not in src
assert 'options_kwargs["skills"] = None' not in src
def test_resolve_sdk_skill_allowlist_empty_when_no_skills_installed(
monkeypatch, tmp_path
):
from backend.apps.skills import skills as skills_mod
monkeypatch.setattr(skills_mod, "SKILLS_DIR", str(tmp_path / "skills"))
assert AgentManager._resolve_sdk_skill_allowlist(None) == []
assert AgentManager._resolve_sdk_skill_allowlist([]) == []
def test_resolve_sdk_skill_allowlist_lists_only_directory_skills(
monkeypatch, tmp_path
):
"""Only `<slug>/SKILL.md` entries count — flat files and dotfiles are
skipped since the CLI's discovery ignores them."""
from backend.apps.skills import skills as skills_mod
skills_root = tmp_path / "skills"
skills_root.mkdir()
(skills_root / "alpha").mkdir()
(skills_root / "alpha" / "SKILL.md").write_text("---\nname: alpha\n---\n")
(skills_root / "beta").mkdir()
(skills_root / "beta" / "SKILL.md").write_text("---\nname: beta\n---\n")
(skills_root / "no-skill-md").mkdir()
(skills_root / "no-skill-md" / "README.md").write_text("not a skill")
(skills_root / "stray.md").write_text("legacy flat file")
(skills_root / ".skills_index.json").write_text("{}")
monkeypatch.setattr(skills_mod, "SKILLS_DIR", str(skills_root))
out = AgentManager._resolve_sdk_skill_allowlist(None)
assert out == ["alpha", "beta"], (
"expected sorted slugs of directories containing SKILL.md, with "
"flat files / dotfiles / partial dirs excluded"
)
def test_resolve_sdk_skill_allowlist_dedups_manually_attached(
monkeypatch, tmp_path
):
"""Skills the user attached via `/` are eagerly injected into the
user message by `_resolve_attached_skills`; they must therefore drop
out of the SDK allowlist for this turn so the same content can't
re-enter via the lazy Skill-tool path."""
from backend.apps.skills import skills as skills_mod
skills_root = tmp_path / "skills"
skills_root.mkdir()
for slug in ("alpha", "beta", "gamma"):
(skills_root / slug).mkdir()
(skills_root / slug / "SKILL.md").write_text(f"---\nname: {slug}\n---\n")
monkeypatch.setattr(skills_mod, "SKILLS_DIR", str(skills_root))
out = AgentManager._resolve_sdk_skill_allowlist(
[{"id": "beta", "name": "Beta"}]
)
assert out == ["alpha", "gamma"]
def test_resolve_sdk_skill_allowlist_excludes_bundled_claude_code_names(
monkeypatch, tmp_path
):
"""The allowlist must never contain bundled Claude Code skill names
just because someone happens to know them — only what's actually on
disk under SKILLS_DIR. This pins the behavior that the bundled
`/init`, `/review`, etc. stay invisible."""
from backend.apps.skills import skills as skills_mod
skills_root = tmp_path / "skills"
skills_root.mkdir()
(skills_root / "user-skill").mkdir()
(skills_root / "user-skill" / "SKILL.md").write_text("---\nname: x\n---\n")
monkeypatch.setattr(skills_mod, "SKILLS_DIR", str(skills_root))
out = AgentManager._resolve_sdk_skill_allowlist(None)
bundled = {
"init",
"review",
"security-review",
"simplify",
"loop",
"schedule",
"update-config",
"keybindings-help",
"fewer-permission-prompts",
"claude-api",
}
assert not bundled.intersection(out)
assert "user-skill" in out
def test_resolve_context_paths_empty_returns_empty():
+316 -8
View File
@@ -1,13 +1,17 @@
"""Smoke tests for /api/skills.
Skills are SKILL.md files synced into `~/.claude/skills`. Tests MUST
NEVER touch the user's real skills dir, so this whole module relies on
the `patched_skills_dir` fixture.
Skills live under `~/.claude/skills/<slug>/SKILL.md` — the layout the
Claude Code CLI auto-discovers (verified against the bundled CLI's path
parser, which expects exactly four segments ending in `SKILL.md`). Tests
MUST NEVER touch the user's real skills dir, so this whole module relies
on the `patched_skills_dir` fixture.
Tests:
- empty list on a fresh dir
- dropping a SKILL.md into the patched dir surfaces in /list
- dropping a `<slug>/SKILL.md` into the patched dir surfaces in /list
- create / get / update / delete round-trip
- frontmatter is stamped on create even when the body has none
- flat-file legacy layout migrates into the directory layout on startup
- workspace/seed writes SKILL.md and meta.json
"""
@@ -24,10 +28,12 @@ def test_list_empty_on_fresh_dir(client, patched_skills_dir):
def test_list_picks_up_dropped_skill(client, patched_skills_dir):
"""Filesystem-driven discovery — dropping a .md file is enough."""
skill_path = os.path.join(patched_skills_dir, "test-skill.md")
"""Filesystem-driven discovery — dropping a `<slug>/SKILL.md` is enough."""
skill_dir = os.path.join(patched_skills_dir, "test-skill")
os.makedirs(skill_dir)
skill_path = os.path.join(skill_dir, "SKILL.md")
with open(skill_path, "w") as f:
f.write("# Test Skill\n\nBody content.\n")
f.write("---\nname: Test Skill\ndescription: a test\n---\n\nBody.\n")
resp = client.get("/api/skills/list")
assert resp.status_code == 200
@@ -35,6 +41,18 @@ def test_list_picks_up_dropped_skill(client, patched_skills_dir):
assert any(s["id"] == "test-skill" for s in skills)
def test_list_ignores_legacy_flat_files(client, patched_skills_dir):
"""A stray `<slug>.md` at the top level isn't a discoverable skill —
the CLI requires the directory layout, so we mirror that."""
flat = os.path.join(patched_skills_dir, "stray.md")
with open(flat, "w") as f:
f.write("body without a home")
resp = client.get("/api/skills/list")
assert resp.status_code == 200
assert resp.json() == {"skills": []}
def test_create_get_update_delete_skill(client, patched_skills_dir):
create = client.post(
"/api/skills/create",
@@ -47,6 +65,9 @@ def test_create_get_update_delete_skill(client, patched_skills_dir):
assert create.status_code == 200, create.text
skill_id = create.json()["skill"]["id"]
skill_md = os.path.join(patched_skills_dir, skill_id, "SKILL.md")
assert os.path.isfile(skill_md), f"expected SKILL.md at {skill_md}"
fetched = client.get(f"/api/skills/{skill_id}")
assert fetched.status_code == 200
assert fetched.json()["name"] == "Smoke"
@@ -58,9 +79,296 @@ def test_create_get_update_delete_skill(client, patched_skills_dir):
assert update.status_code == 200
assert update.json()["skill"]["description"] == "smoke v2"
with open(skill_md, encoding="utf-8") as f:
on_disk = f.read()
assert "description: smoke v2" in on_disk, (
"metadata-only update should patch SKILL.md frontmatter so the CLI "
"listing stays in sync"
)
deleted = client.delete(f"/api/skills/{skill_id}")
assert deleted.status_code == 200
assert not os.path.exists(os.path.join(patched_skills_dir, f"{skill_id}.md"))
assert not os.path.exists(os.path.join(patched_skills_dir, skill_id))
def test_create_stamps_frontmatter_when_body_has_none(client, patched_skills_dir):
"""Without frontmatter the CLI silently drops a skill from the listing,
so the create endpoint MUST emit a YAML block even if the caller sent
a plain Markdown body."""
create = client.post(
"/api/skills/create",
json={
"name": "Bare",
"description": "no fm in body",
"content": "Just a body, no frontmatter.\n",
},
)
assert create.status_code == 200, create.text
skill_id = create.json()["skill"]["id"]
skill_md = os.path.join(patched_skills_dir, skill_id, "SKILL.md")
with open(skill_md, encoding="utf-8") as f:
on_disk = f.read()
assert on_disk.startswith("---\n")
assert "name: Bare" in on_disk
assert "description: no fm in body" in on_disk
assert "Just a body, no frontmatter." in on_disk
def test_inline_frontmatter_edits_in_body_are_honored(client, patched_skills_dir):
"""Editing the `---` block inside the content textarea must stick when
the caller doesn't separately override via the form fields.
Previously `update_skill` always stamped the file with `meta`'s
pre-existing name/description, silently discarding the user's inline
YAML edits. New precedence: explicit form field > inline frontmatter
> previous index value. This test exercises the inline-only path.
"""
create = client.post(
"/api/skills/create",
json={"name": "Original", "description": "old desc", "content": "body"},
)
assert create.status_code == 200, create.text
skill_id = create.json()["skill"]["id"]
new_body = (
"---\n"
"name: EditedInline\n"
"description: edited inline\n"
"---\n\n"
"fresh body\n"
)
resp = client.put(f"/api/skills/{skill_id}", json={"content": new_body})
assert resp.status_code == 200, resp.text
listed = client.get("/api/skills/list").json()["skills"]
surfaced = next(s for s in listed if s["id"] == skill_id)
assert surfaced["name"] == "EditedInline", (
"inline frontmatter `name:` must beat stale index value when the "
"PUT body doesn't include an explicit `name` field"
)
assert surfaced["description"] == "edited inline"
def test_form_name_still_beats_inline_frontmatter(client, patched_skills_dir):
"""The opposite direction of the precedence rule: when the caller
sends BOTH an explicit `name` form field AND inline frontmatter, the
form field wins. This is the typical frontend save flow."""
create = client.post(
"/api/skills/create",
json={"name": "Original", "description": "old", "content": "body"},
)
skill_id = create.json()["skill"]["id"]
conflicting_body = (
"---\n"
"name: FromInline\n"
"description: from inline\n"
"---\n\n"
"body\n"
)
resp = client.put(
f"/api/skills/{skill_id}",
json={"name": "FromForm", "content": conflicting_body},
)
assert resp.status_code == 200
listed = client.get("/api/skills/list").json()["skills"]
surfaced = next(s for s in listed if s["id"] == skill_id)
assert surfaced["name"] == "FromForm", (
"explicit form `name` must beat inline frontmatter when both "
"are supplied"
)
# description wasn't on the form, so inline wins
assert surfaced["description"] == "from inline"
def test_frontmatter_roundtrips_special_characters(client, patched_skills_dir):
"""A description with embedded quotes / colons / hashes must survive
arbitrary save-load-save cycles unchanged. The previous parser blindly
stripped one pair of outer quotes without un-escaping `\\"`, so each
save accumulated more backslashes until the description was a wall of
`\\\\\\\\\\"`. Regression test for that bug.
"""
tricky = 'has "quotes": and # hash'
create = client.post(
"/api/skills/create",
json={"name": "Tricky", "description": tricky, "content": "body\n"},
)
assert create.status_code == 200, create.text
skill_id = create.json()["skill"]["id"]
# Three metadata-only saves: every one re-stamps frontmatter, so a
# lossy decoder would compound the damage. The list endpoint reads
# the description back from disk via the frontmatter parser.
for _ in range(3):
resp = client.put(f"/api/skills/{skill_id}", json={"description": tricky})
assert resp.status_code == 200
listed = client.get("/api/skills/list").json()["skills"]
surfaced = next(s for s in listed if s["id"] == skill_id)
assert surfaced["description"] == tricky, (
"description with quotes/colons/hash must round-trip exactly across "
"successive saves; lossy parser would accumulate backslashes"
)
def test_migration_moves_flat_file_into_directory(patched_skills_dir, monkeypatch):
"""Pre-existing `<slug>.md` files (from before the directory-layout
migration) must be relocated to `<slug>/SKILL.md` exactly once at
startup so the CLI's discovery picks them up."""
from backend.apps.skills import skills as skills_mod
flat = os.path.join(patched_skills_dir, "legacy.md")
with open(flat, "w") as f:
f.write("# Legacy\n\nbody.")
skills_mod._save_index({"legacy": {"name": "Legacy", "description": "old"}})
skills_mod._migrate_flat_to_dir_layout()
assert not os.path.exists(flat), "flat file should be removed after migration"
new_path = os.path.join(patched_skills_dir, "legacy", "SKILL.md")
assert os.path.isfile(new_path), f"expected migrated SKILL.md at {new_path}"
with open(new_path, encoding="utf-8") as f:
migrated = f.read()
assert migrated.startswith("---\n"), "migration should stamp frontmatter"
assert "name: Legacy" in migrated
assert "description: old" in migrated
assert "body." in migrated
# Idempotent: a second run must not crash or duplicate work.
skills_mod._migrate_flat_to_dir_layout()
assert os.path.isfile(new_path)
def test_migration_preserves_existing_directory(patched_skills_dir):
"""If a directory already exists for a slug, the flat file is removed
(the dir wins) — never overwritten."""
from backend.apps.skills import skills as skills_mod
skill_dir = os.path.join(patched_skills_dir, "shared")
os.makedirs(skill_dir)
canonical = os.path.join(skill_dir, "SKILL.md")
with open(canonical, "w") as f:
f.write("---\nname: shared\n---\n\ncanonical body")
flat = os.path.join(patched_skills_dir, "shared.md")
with open(flat, "w") as f:
f.write("legacy body that should not win")
skills_mod._migrate_flat_to_dir_layout()
assert not os.path.exists(flat)
with open(canonical, encoding="utf-8") as f:
assert "canonical body" in f.read()
def test_installed_slugs_cache_avoids_rescanning(patched_skills_dir, monkeypatch):
"""`get_installed_slugs` must hit the filesystem on the first call,
then serve subsequent calls from cache without re-walking SKILLS_DIR.
The agent loop calls this on every turn, so a per-turn rescan would
cost N*T syscalls in the hot path for N skills * T turns."""
from backend.apps.skills import skills as skills_mod
skill_dir = os.path.join(patched_skills_dir, "alpha")
os.makedirs(skill_dir)
with open(os.path.join(skill_dir, "SKILL.md"), "w") as f:
f.write("---\nname: alpha\n---\n")
real_listdir = os.listdir
call_count = {"n": 0}
def counting_listdir(path):
if os.path.abspath(path) == os.path.abspath(patched_skills_dir):
call_count["n"] += 1
return real_listdir(path)
monkeypatch.setattr(skills_mod.os, "listdir", counting_listdir)
skills_mod._invalidate_installed_slugs_cache()
assert skills_mod.get_installed_slugs() == ["alpha"]
first = call_count["n"]
assert first == 1, "first call must rescan"
for _ in range(5):
assert skills_mod.get_installed_slugs() == ["alpha"]
assert call_count["n"] == first, (
"subsequent calls must be served from cache without re-listing "
"SKILLS_DIR"
)
def test_installed_slugs_cache_invalidated_on_create(client, patched_skills_dir):
"""Creating a skill via /api/skills/create must invalidate the cache
so the next allowlist query reflects the new slug. Without this the
next agent turn would silently fail to expose the freshly-installed
skill to the model."""
from backend.apps.skills import skills as skills_mod
assert skills_mod.get_installed_slugs() == []
resp = client.post(
"/api/skills/create",
json={
"name": "FreshSkill",
"description": "test",
"content": "body",
},
)
assert resp.status_code == 200
skill_id = resp.json()["skill"]["id"]
assert skill_id in skills_mod.get_installed_slugs(), (
"newly-created slug should appear without an explicit "
"cache-bust call from the test"
)
def test_installed_slugs_cache_invalidated_on_delete(client, patched_skills_dir):
"""Deleting a skill must invalidate the cache so the slug drops out
of the next allowlist query. Otherwise the agent would keep being
told about a skill the user just removed, and a `Skill(slug)` call
would fail in the CLI with a confusing 'skill not found' error."""
from backend.apps.skills import skills as skills_mod
client.post(
"/api/skills/create",
json={"name": "Doomed", "description": "x", "content": "body"},
)
assert "doomed" in skills_mod.get_installed_slugs()
resp = client.delete("/api/skills/doomed")
assert resp.status_code == 200
assert "doomed" not in skills_mod.get_installed_slugs(), (
"deleted slug must drop out of the cached list immediately"
)
def test_installed_slugs_cache_keyed_by_skills_dir(monkeypatch, tmp_path):
"""The cache is keyed by SKILLS_DIR so a test that monkeypatches
that dir gets a fresh scan automatically — important because pytest
re-uses the same module-scoped cache across all tests, and we don't
want test A's skills bleeding into test B's view."""
from backend.apps.skills import skills as skills_mod
dir_a = tmp_path / "a"
dir_a.mkdir()
(dir_a / "only-in-a").mkdir()
(dir_a / "only-in-a" / "SKILL.md").write_text("---\nname: a\n---\n")
dir_b = tmp_path / "b"
dir_b.mkdir()
(dir_b / "only-in-b").mkdir()
(dir_b / "only-in-b" / "SKILL.md").write_text("---\nname: b\n---\n")
monkeypatch.setattr(skills_mod, "SKILLS_DIR", str(dir_a))
skills_mod._invalidate_installed_slugs_cache()
assert skills_mod.get_installed_slugs() == ["only-in-a"]
monkeypatch.setattr(skills_mod, "SKILLS_DIR", str(dir_b))
assert skills_mod.get_installed_slugs() == ["only-in-b"], (
"switching SKILLS_DIR must implicitly invalidate the cache "
"without an explicit bust"
)
def test_workspace_seed_writes_files(client, patched_skills_dir):