tools/skills: list, read, and search skills

skill_list shows what was discovered (name, source root, directory, description)
together with the copies a name collision shadowed and the notes from repaired
frontmatter, so a third-party skill never disappears quietly. skill_read
addresses a skill by name and defaults to its SKILL.md, reusing read_file for
ranges, line numbers, and the continuation token; a bundled file is resolved
inside the skill directory only, symlinks included. skill_search greps every
visible skill (or one) and rebases each hit on its skill, which makes a match
attributable without absolute paths.

Reads need no grant: the host registers the roots as static read-only roots on
the workspace policy, so read_file/grep_files see skill directories too while
write access stays denied — skill writes will go through their own confirm.
Grep used to drop every file outside the primary workspace root, which also hid
hits under a granted directory; those now come back as absolute paths.
This commit is contained in:
2026-09-22 22:42:45 +08:00
parent a9f58428bf
commit 1d05da62c2
12 changed files with 599 additions and 3 deletions
+56
View File
@@ -34,6 +34,7 @@ _TOOL_RESULT_PREVIEW = 120
_TOOL_ARGS_PREVIEW = 80
_ASK_SUBJECT_PREVIEW = 60
_MCP_TOOL_PREFIX = "mcp__"
_SKILL_MAIN_FILE = "SKILL.md"
# Process/session display flags (set from ReplState / slash).
_verbose_tool_results: ContextVar[bool] = ContextVar("verbose_tool_results", default=False)
@@ -701,6 +702,58 @@ def _mcp_detail(_args_json: str, result: str) -> str:
return _pretty_segments(("(done)", "green"))
def _skill_list_prefix(_args_json: str) -> str:
return _pretty_prefix("* Skills ")
def _skill_list_detail(_args_json: str, result: str) -> str:
"""``skill_list`` outcome: how many skills were found, or an error."""
if result.startswith("error:"):
return _pretty_segments((f"({result})", "red"))
match = re.match(r"skills: (\d+)", result)
if match is None:
return _pretty_segments(("(none)", "dim"))
count = int(match.group(1))
unit = "skill" if count == 1 else "skills"
return _pretty_segments((f"({count} {unit})", None if count else "dim"))
def _skill_read_prefix(args_json: str) -> str:
name = _json_str_arg(args_json, "name") or "?"
file = _json_str_arg(args_json, "file") or ""
subject = f"'{name}'" if file in {"", _SKILL_MAIN_FILE} else f"'{name}' {file}"
return _pretty_prefix(f"* Skill {subject} ")
def _skill_read_detail(_args_json: str, result: str) -> str:
"""``skill_read`` outcome: the range read, or the failure kind."""
if result.startswith("error:"):
return _pretty_segments((f"({result})", "red"))
return _range_detail(result) or _pretty_segments(("(done)", "green"))
def _skill_search_prefix(args_json: str) -> str:
pattern = _json_str_arg(args_json, "pattern") or "?"
scope = _json_str_arg(args_json, "skill") or ""
where = f" in '{scope}'" if scope else ""
return _pretty_prefix(f"* Skill Search '{_preview(pattern, _ASK_SUBJECT_PREVIEW)}'{where} ")
def _skill_search_detail(_args_json: str, result: str) -> str:
"""``skill_search`` outcome: matches and skills, none, or an error."""
if result.startswith("error:"):
return _pretty_segments((f"({result})", "red"))
if result == "(no matches)":
return _pretty_segments(("(no matches)", "dim"))
if result == "(no skills to search)":
return _pretty_segments(("(no skills)", "dim"))
matches = [line for line in result.splitlines() if line and not line.startswith("...[truncated")]
skills = {line.split("/", 1)[0] for line in matches}
match_unit = "match" if len(matches) == 1 else "matches"
skill_unit = "skill" if len(skills) == 1 else "skills"
return _pretty_segments((f"({len(matches)} {match_unit} in {len(skills)} {skill_unit})", None))
# Pretty tools: ``prefix`` prints as the call starts, ``detail`` when it lands.
_PRETTY_LINES: dict[str, PrettyLine] = {
"read_file": PrettyLine(prefix=_read_file_prefix, detail=_read_file_detail),
@@ -725,6 +778,9 @@ _PRETTY_LINES: dict[str, PrettyLine] = {
"delete_path": _mutator_line("delete_path"),
"new_temporary_workspace": PrettyLine(prefix=_temp_dir_prefix, detail=_temp_dir_detail),
"request_directory_access": PrettyLine(prefix=_access_prefix, detail=_access_detail),
"skill_list": PrettyLine(prefix=_skill_list_prefix, detail=_skill_list_detail),
"skill_read": PrettyLine(prefix=_skill_read_prefix, detail=_skill_read_detail),
"skill_search": PrettyLine(prefix=_skill_search_prefix, detail=_skill_search_detail),
"open_pty": PrettyLine(prefix=_open_pty_prefix, detail=_open_pty_detail),
"read_pty": _pty_session_line("Read", _read_pty_detail),
"write_pty": _pty_session_line("Write", _pty_write_detail),
+1
View File
@@ -5,6 +5,7 @@ from .discovery import Skill as Skill
from .discovery import SkillRoot as SkillRoot
from .discovery import SkillStore as SkillStore
from .discovery import build_roots as build_roots
from .discovery import resolve_skill_file as resolve_skill_file
from .discovery import validate_skill_name as validate_skill_name
from .frontmatter import FrontmatterError as FrontmatterError
from .frontmatter import parse_frontmatter as parse_frontmatter
+20
View File
@@ -136,6 +136,26 @@ def validate_skill_name(name: str) -> str | None:
return None
def resolve_skill_file(skill: Skill, relative: str) -> Path | None:
"""Resolve *relative* inside *skill*; ``None`` when it escapes the directory.
The default (empty *relative*) is the skill's own ``SKILL.md``. Symlinks are
followed by the resolve, so a link pointing outside the skill is rejected
instead of becoming a way around the boundary.
"""
token = relative.strip() or SKILL_FILE
candidate = Path(token)
if candidate.is_absolute():
return None
try:
base = skill.path.resolve()
target = (base / candidate).resolve()
_ = target.relative_to(base)
except OSError, ValueError:
return None
return target
class SkillStore:
"""Skills discovered under *roots*, highest-priority root first."""
+8
View File
@@ -60,6 +60,13 @@ from .process import run_argv as run_argv
from .process import run_argv_batch as run_argv_batch
from .process import write_pty as write_pty
from .process import write_pty_keys as write_pty_keys
from .skills import SKILL_TOOL_NAMES as SKILL_TOOL_NAMES
from .skills import SKILL_TOOLS as SKILL_TOOLS
from .skills import format_skill_table as format_skill_table
from .skills import get_skill_store as get_skill_store
from .skills import skill_list as skill_list
from .skills import skill_read as skill_read
from .skills import skill_search as skill_search
from .temp_workspace import cleanup_temporary_workspaces as cleanup_temporary_workspaces
from .temp_workspace import new_temporary_workspace as new_temporary_workspace
from .todo import TODO_TOOLS as TODO_TOOLS
@@ -106,5 +113,6 @@ from .workspace import set_command_denylist as set_command_denylist
from .workspace import set_path_denylist as set_path_denylist
from .workspace import set_policy_confirm_hook as set_policy_confirm_hook
from .workspace import set_policy_confirm_timeout as set_policy_confirm_timeout
from .workspace import set_static_read_roots as set_static_read_roots
from .workspace import set_workspace_root as set_workspace_root
from .workspace import within_workspace_roots as within_workspace_roots
+4
View File
@@ -223,6 +223,7 @@ def _ensure_builtin_definitions_registered(catalog: ToolCatalog) -> None:
from plyngent.tools.file import FILE_TOOLS
from plyngent.tools.net import NET_TOOLS
from plyngent.tools.process import PROCESS_TOOLS
from plyngent.tools.skills import SKILL_TOOLS
from plyngent.tools.todo import TODO_TOOLS
from plyngent.tools.vcs import VCS_TOOLS
@@ -233,6 +234,7 @@ def _ensure_builtin_definitions_registered(catalog: ToolCatalog) -> None:
*VCS_TOOLS,
*CHAT_TOOLS,
*TODO_TOOLS,
*SKILL_TOOLS,
*NET_TOOLS,
):
if catalog.get(definition.name) is None:
@@ -254,6 +256,7 @@ def register_builtin_tools(*, force: bool = False) -> ToolCatalog:
import plyngent.tools.file as _file_tools
import plyngent.tools.net as _net_tools
import plyngent.tools.process as _process_tools
import plyngent.tools.skills as _skills_tools
import plyngent.tools.temp_workspace as _temp_workspace_tools
import plyngent.tools.todo as _todo_tools
import plyngent.tools.vcs as _vcs_tools
@@ -263,6 +266,7 @@ def register_builtin_tools(*, force: bool = False) -> ToolCatalog:
_file_tools,
_net_tools,
_process_tools,
_skills_tools,
_temp_workspace_tools,
_todo_tools,
_vcs_tools,
+3
View File
@@ -14,6 +14,7 @@ if TYPE_CHECKING:
from pathlib import Path
from plyngent.agent.todo_stack import TodoStack
from plyngent.skills import SkillStore
from plyngent.tools.workspace import AccessMode, WorkspacePolicy
@@ -40,6 +41,8 @@ class InstanceState:
# Path/command policy bag for this host (preferred over process globals).
workspace: WorkspacePolicy = field(default_factory=_default_workspace_policy)
data: PersistentDataView[Any] = field(default_factory=_default_instance_data)
# Discovered skills for this host (None when the feature is off).
skills: SkillStore | None = None
# Ephemeral process maps (PTY etc.) may hang here later.
extras: dict[str, Any] = field(default_factory=dict)
+3 -1
View File
@@ -103,7 +103,9 @@ def _match_lines(
try:
rel = str(file_path.resolve().relative_to(root))
except ValueError:
return
# Outside the primary root (a grant / temp workspace / skill directory):
# keep the absolute path so the hit stays addressable.
rel = str(file_path.resolve())
found = 0
for line_no, line in enumerate(text.splitlines(), start=1):
if found >= remaining:
+217
View File
@@ -0,0 +1,217 @@
"""Skill tools: discover, read, search, create, and edit skill directories.
A skill is a directory holding a ``SKILL.md`` (frontmatter + Markdown body) plus
optional bundled files — scripts, references, templates. Discovery, parsing, and
path rules live in :mod:`plyngent.skills`; this module is the model-facing shell.
Reading needs no grant: the host registers the roots as static read-only roots,
so ``read_file`` / ``grep_files`` see them too. Writing goes through the skill
tools' own grant (a per-call confirm, or ``[skills].allow_write``) and is always
confined to a skill root — that is why these tools never inherit the
directory-access path of ``request_directory_access``.
"""
from __future__ import annotations
from plyngent.agent import ToolTag, tool
from plyngent.agent.budget import DEFAULT_TOOL_RESULT_MAX_CHARS
from plyngent.skills import SKILL_FILE, Skill, SkillStore, resolve_skill_file
from plyngent.tools.file.grep_files import DEFAULT_MAX_MATCHES
from plyngent.tools.file.grep_files import grep_files as _grep_files
from plyngent.tools.file.read import read_file as _read_file
# Keeps a listing readable when a third-party root holds a pile of skills.
_MAX_LISTED_ISSUES = 8
def get_skill_store() -> SkillStore | None:
"""The skill store bound on the host instance, if any."""
from plyngent.tools.context import get_instance
instance = get_instance()
return instance.skills if instance is not None else None
def _store_or_error() -> SkillStore | str:
store = get_skill_store()
if store is None:
return "error: skills are not available in this session"
return store
def format_skill_table(store: SkillStore) -> str:
"""Model- and human-facing listing: visible skills, shadows, roots, notes.
Shadowed copies and per-skill notes are printed rather than hidden: a skill
that lost a name collision, or whose frontmatter had to be repaired, must
stay visible to whoever can fix it.
"""
skills = store.skills
lines = [f"skills: {len(skills)}"]
width = max((len(skill.name) for skill in skills), default=0)
for skill in skills:
description = skill.description or "(no description)"
lines.append(f"{skill.name.ljust(width)} [{skill.root.label}] {skill.path} — {description}")
shadows = store.shadowed()
if shadows:
parts: list[str] = []
for name, copies in shadows.items():
others = ", ".join(copy.root.label for copy in copies[1:])
parts.append(f"{name} — {copies[0].root.label} wins (also {others})")
lines.append("shadowed: " + "; ".join(parts))
notes = [f"{skill.name}: {issue}" for skill in skills for issue in skill.issues]
if notes:
lines.append("notes: " + "; ".join(notes[:_MAX_LISTED_ISSUES]))
if len(notes) > _MAX_LISTED_ISSUES:
lines.append(f"notes: … and {len(notes) - _MAX_LISTED_ISSUES} more")
absent = [root for root in store.roots if not root.path.is_dir()]
if absent:
lines.append("roots without a directory: " + ", ".join(f"{root.path} ({root.label})" for root in absent))
return "\n".join(lines)
def _unknown_skill(name: str) -> str:
return f"error: unknown skill {name!r}; call skill_list to see the discovered skills"
@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.READ_ONLY)
async def skill_list() -> str:
"""List discovered skills: name, source root, directory, and description.
Skills come from the roots in ``[skills]`` (our own user directory and other
harnesses' skill directories by default); the first copy of a name wins and
the copies it shadows are listed too. A skill's ``SKILL.md`` is the
instruction set — read one with ``skill_read`` before following it, and use
``skill_search`` to find a skill by its content.
"""
store = _store_or_error()
if isinstance(store, str):
return store
return format_skill_table(store)
@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.READ_ONLY)
async def skill_read(
name: str,
file: str = SKILL_FILE,
*,
offset: int = 0,
limit: int | None = None,
with_lineno: bool = False,
max_chars: int = DEFAULT_TOOL_RESULT_MAX_CHARS,
) -> str:
"""Read a skill's ``SKILL.md`` (default) or one of its bundled files.
``name`` is the skill name from ``skill_list``; ``file`` is a path inside
that skill (a script or reference it mentions). ``offset`` is a 0-based line
start, ``with_lineno`` prefixes 1-based line numbers, and ``max_chars`` caps
the slice with a ``truncate_token`` for ``get_truncated`` to continue.
"""
store = _store_or_error()
if isinstance(store, str):
return store
skill = store.find(name)
if skill is None:
return _unknown_skill(name)
target = resolve_skill_file(skill, file)
if target is None:
return f"error: file {file!r} is outside skill {skill.name!r}"
if not target.is_file():
return f"error: no such file in skill {skill.name!r}: {file}"
return await _read_file.handler(
str(target),
offset=offset,
limit=limit,
with_lineno=with_lineno,
max_chars=max_chars,
)
def _skill_hits(skill: Skill, result: str) -> list[str]:
"""Attribute one skill's grep lines to that skill.
Hits outside the workspace root come back as absolute paths, so the file
part is rebased on the skill directory before the skill name is prepended.
"""
base = str(skill.path.resolve())
hits: list[str] = []
for line in result.splitlines():
if not line or line.startswith("...[truncated"):
continue
path, sep, rest = line.partition(":")
relative = path[len(base) :].lstrip("/\\") if path.startswith(base) else path
hits.append(f"{skill.name}/{relative}{sep}{rest}")
return hits
def _search_targets(store: SkillStore, skill: str) -> list[Skill] | str:
"""Every visible skill, or the single named one."""
if not skill.strip():
return store.skills
found = store.find(skill)
if found is None:
return _unknown_skill(skill)
return [found]
@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.READ_ONLY)
async def skill_search(
pattern: str,
skill: str = "",
*,
case_insensitive: bool = False,
max_matches: int = DEFAULT_MAX_MATCHES,
) -> str:
"""Search skill contents with a regular expression.
Returns ``<skill>/<file>:<line>: content`` lines across every visible skill
(or only *skill* when given), covering ``SKILL.md`` and any bundled files.
Use ``skill_list`` for names and descriptions, and ``glob_paths`` on a
skill's directory when you are looking for a file rather than its content.
"""
store = _store_or_error()
if isinstance(store, str):
return store
targets = _search_targets(store, skill)
if isinstance(targets, str):
return targets
if not targets:
return "(no skills to search)"
hits: list[str] = []
for target in targets:
remaining = max_matches - len(hits)
if remaining <= 0:
break
result = await _grep_files.handler(
pattern,
str(target.path),
case_insensitive=case_insensitive,
max_matches=remaining,
)
if result.startswith("error:") or result == "(no matches)":
if result.startswith("error:"):
return result
continue
hits.extend(_skill_hits(target, result))
if not hits:
return "(no matches)"
body = "\n".join(hits[:max_matches])
if len(hits) >= max_matches:
body += f"\n...[truncated at {max_matches} matches]"
return body
SKILL_TOOLS = [
skill_list,
skill_read,
skill_search,
]
# Fed to the catalog selection filter: ``[skills].enabled = false`` drops these
# from the model-visible registry (definitions still exist in the catalog).
SKILL_TOOL_NAMES = frozenset(definition.name for definition in SKILL_TOOLS)
+32 -1
View File
@@ -98,6 +98,10 @@ class WorkspacePolicy:
config_allow: dict[Path, AccessMode] = field(default_factory=dict)
# Process-scoped YOLO / --yes grants (resolved path → mode); never persisted.
yolo_allow: dict[Path, AccessMode] = field(default_factory=dict)
# Host-contributed read-only roots (skill directories): every path tool may
# read them, nothing may write them. Not a grant, so they stay out of
# ``/grants``; writes go through the owning tool's own confirm.
static_read: list[Path] = field(default_factory=list)
def _bound_instance() -> InstanceState | None:
@@ -285,7 +289,7 @@ def _under_any_root(resolved: Path, instance: InstanceState, policy: WorkspacePo
def granted_access_mode(resolved: Path) -> AccessMode | None:
"""Highest config / session / process mode covering *resolved*, if any."""
"""Highest config / session / process / static-read mode covering *resolved*."""
policy = active_workspace_policy()
session = _bound_session()
buckets: list[Mapping[Path, AccessMode]] = [policy.config_allow, policy.yolo_allow]
@@ -296,9 +300,36 @@ def granted_access_mode(resolved: Path) -> AccessMode | None:
for root, granted in grants.items():
if (resolved == root or resolved.is_relative_to(root)) and (best is None or granted > best):
best = granted
for root in policy.static_read:
if (resolved == root or resolved.is_relative_to(root)) and (best is None or best < AccessMode.READ):
best = AccessMode.READ
return best
def set_static_read_roots(
roots: Sequence[Path | str],
*,
instance: InstanceState | None = None,
) -> list[Path]:
"""Install host-contributed read-only roots (resolved); returns them.
Replaces any previous list. Roots that are not existing directories are
dropped: an absent skill directory holds nothing to read, and a missing
entry must never turn into an implicit write permission later.
"""
inst = instance if instance is not None else require_bound_instance()
resolved: list[Path] = []
for root in roots:
try:
path = Path(root).expanduser().resolve()
except OSError:
continue
if path.is_dir():
resolved.append(path)
inst.workspace.static_read[:] = resolved
return resolved
def _path_grant_covers(resolved: Path, required: AccessMode) -> bool:
"""Whether a config / session / process grant covers *resolved*."""
granted = granted_access_mode(resolved)
+47
View File
@@ -821,6 +821,53 @@ def test_every_builtin_tool_has_a_pretty_line() -> None:
assert missing == []
async def test_pretty_skill_tools(capsys: pytest.CaptureFixture[str]) -> None:
await render_events(
_aiter(
[_pretty_call("skill_list", "{}"), _result("skills: 3\npdf-processing [plyngent-user] /x — Extract PDFs")]
)
)
out = capsys.readouterr().out
assert "* Skills (3 skills)" in out
await render_events(
_aiter(
[
_pretty_call("skill_read", '{"name": "pdf-processing"}'),
_result("L1-40\n# PDF processing\n"),
]
)
)
out = capsys.readouterr().out
assert "* Skill 'pdf-processing' L1-40 (done)" in out
await render_events(
_aiter(
[
_pretty_call("skill_read", '{"name": "pdf-processing", "file": "scripts/run.sh"}'),
_result("L1-3\n#!/bin/sh\n"),
]
)
)
out = capsys.readouterr().out
assert "* Skill 'pdf-processing' scripts/run.sh L1-3 (done)" in out
await render_events(
_aiter(
[
_pretty_call("skill_search", '{"pattern": "pdftotext"}'),
_result("alpha/SKILL.md:3: Use pdftotext.\nbeta/notes.md:1: pdftotext again"),
]
)
)
out = capsys.readouterr().out
assert "* Skill Search 'pdftotext' (2 matches in 2 skills)" in out
await render_events(_aiter([_pretty_call("skill_search", '{"pattern": "nope"}'), _result("(no matches)")]))
out = capsys.readouterr().out
assert "* Skill Search 'nope' (no matches)" in out
async def test_pretty_prefix_shown_while_call_runs(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
+14 -1
View File
@@ -12,6 +12,7 @@ from plyngent.tools.chat import CHAT_TOOLS
from plyngent.tools.file import FILE_TOOLS
from plyngent.tools.net import NET_TOOLS
from plyngent.tools.process import PROCESS_TOOLS
from plyngent.tools.skills import SKILL_TOOLS
from plyngent.tools.todo import TODO_TOOLS
from plyngent.tools.vcs import VCS_TOOLS
@@ -19,7 +20,16 @@ from plyngent.tools.vcs import VCS_TOOLS
def test_default_tool_names_match_group_lists() -> None:
register_builtin_tools()
selected = default_tool_definitions(surface="local")
groups = [*ACCESS_TOOLS, *FILE_TOOLS, *PROCESS_TOOLS, *VCS_TOOLS, *CHAT_TOOLS, *TODO_TOOLS, *NET_TOOLS]
groups = [
*ACCESS_TOOLS,
*FILE_TOOLS,
*PROCESS_TOOLS,
*VCS_TOOLS,
*CHAT_TOOLS,
*TODO_TOOLS,
*SKILL_TOOLS,
*NET_TOOLS,
]
assert sorted(t.name for t in selected) == sorted(t.name for t in groups)
assert len(selected) == len(groups)
@@ -65,6 +75,9 @@ def test_read_only_classification() -> None:
"vcs_log",
"vcs_branch",
"fetch",
"skill_list",
"skill_read",
"skill_search",
}
+194
View File
@@ -0,0 +1,194 @@
from __future__ import annotations
from contextlib import contextmanager
from typing import TYPE_CHECKING
from plyngent.config.models import SkillsConfig
from plyngent.skills import SKILL_FILE, SkillStore, build_roots
from plyngent.tools.context import InstanceState, bind_instance
from plyngent.tools.skills import SKILL_TOOL_NAMES, format_skill_table, skill_list, skill_read, skill_search
from plyngent.tools.workspace import set_static_read_roots, set_workspace_root
if TYPE_CHECKING:
from collections.abc import Generator
from pathlib import Path
def _write_skill(root: Path, name: str, *, description: str | None = "A test skill", body: str = "Body.") -> Path:
directory = root / name
directory.mkdir(parents=True, exist_ok=True)
front = "---\n"
if description is not None:
front += f"name: {name}\ndescription: {description}\n"
front += "---\n\n"
_ = (directory / SKILL_FILE).write_text(front + body + "\n", encoding="utf-8")
return directory
@contextmanager
def _bound(tmp_path: Path, *roots: Path, skills: SkillStore | None = None) -> Generator[InstanceState]:
"""Bind an instance whose skill store scans *roots*, like the CLI does."""
workspace = tmp_path / "project"
workspace.mkdir(exist_ok=True)
store = (
skills
if skills is not None
else SkillStore(
build_roots(
SkillsConfig(paths=[str(r) for r in roots], discover=[]), workspace=workspace, user_config_dir=tmp_path
)
)
)
instance = InstanceState(workspace_root=workspace, skills=store)
with bind_instance(instance):
_ = set_workspace_root(workspace)
_ = set_static_read_roots(roots)
yield instance
async def test_skill_list_without_a_store_returns_an_error(tmp_path: Path) -> None:
with _bound(tmp_path, skills=None) as instance:
instance.skills = None
assert await skill_list.handler() == "error: skills are not available in this session"
async def test_skill_list_reports_skills_roots_and_absent_dirs(tmp_path: Path) -> None:
root = tmp_path / "team"
_ = _write_skill(root, "acme-deploy", description="Deploy the acme stack")
missing = tmp_path / "missing"
with _bound(tmp_path, root, missing):
out = await skill_list.handler()
assert "skills: 1" in out
assert "acme-deploy [config-user]" in out
assert "Deploy the acme stack" in out
assert str(root / "acme-deploy") in out
assert "roots without a directory" in out
assert str(missing) in out
async def test_skill_list_marks_shadowed_copies(tmp_path: Path) -> None:
high = tmp_path / "team"
low = tmp_path / "personal"
_ = _write_skill(high, "shared", description="Team copy")
_ = _write_skill(low, "shared", description="Personal copy")
with _bound(tmp_path, high, low):
out = await skill_list.handler()
assert "Team copy" in out
assert "Personal copy" not in out
assert "shadowed: shared — config-user wins (also config-user)" in out
async def test_skill_list_shows_notes(tmp_path: Path) -> None:
root = tmp_path / "team"
directory = root / "no-desc"
directory.mkdir(parents=True)
_ = (directory / SKILL_FILE).write_text("# Heading\n\nBody paragraph.\n", encoding="utf-8")
with _bound(tmp_path, root):
out = await skill_list.handler()
assert "notes: no-desc: no frontmatter description; using the first paragraph" in out
async def test_skill_read_defaults_to_skill_md(tmp_path: Path) -> None:
root = tmp_path / "team"
_ = _write_skill(root, "acme", body="Step one.")
with _bound(tmp_path, root):
out = await skill_read.handler("acme")
assert out.startswith("L1-")
assert "Step one." in out
async def test_skill_read_bundled_file_with_line_numbers(tmp_path: Path) -> None:
root = tmp_path / "team"
directory = _write_skill(root, "acme")
(directory / "scripts").mkdir()
_ = (directory / "scripts" / "run.sh").write_text("#!/bin/sh\necho hi\n", encoding="utf-8")
with _bound(tmp_path, root):
out = await skill_read.handler("acme", "scripts/run.sh", with_lineno=True)
assert " 1|#!/bin/sh" in out
async def test_skill_read_refuses_to_escape_the_skill(tmp_path: Path) -> None:
root = tmp_path / "team"
_ = _write_skill(root, "acme")
_ = (tmp_path / "outside.txt").write_text("not yours\n", encoding="utf-8")
with _bound(tmp_path, root):
assert await skill_read.handler("acme", "../outside.txt") == (
"error: file '../outside.txt' is outside skill 'acme'"
)
assert await skill_read.handler("acme", "/etc/hostname") == (
"error: file '/etc/hostname' is outside skill 'acme'"
)
async def test_skill_read_unknown_skill_and_missing_file(tmp_path: Path) -> None:
root = tmp_path / "team"
_ = _write_skill(root, "acme")
with _bound(tmp_path, root):
assert (
await skill_read.handler("nope")
== "error: unknown skill 'nope'; call skill_list to see the discovered skills"
)
assert await skill_read.handler("acme", "missing.md") == "error: no such file in skill 'acme': missing.md"
async def test_skill_read_truncates_with_a_continuation_token(tmp_path: Path) -> None:
root = tmp_path / "team"
_ = _write_skill(root, "acme", body="x" * 500)
with _bound(tmp_path, root):
out = await skill_read.handler("acme", max_chars=120)
assert "truncate_token=" in out
async def test_skill_search_attributes_hits_to_their_skill(tmp_path: Path) -> None:
root = tmp_path / "team"
_ = _write_skill(root, "alpha", body="Use pdftotext for extraction.")
beta = _write_skill(root, "beta", body="No match in the body.")
_ = (beta / "notes.md").write_text("pdftotext again\n", encoding="utf-8")
with _bound(tmp_path, root):
out = await skill_search.handler("pdftotext")
assert "alpha/SKILL.md:" in out
assert "beta/notes.md:" in out
async def test_skill_search_single_skill_and_errors(tmp_path: Path) -> None:
root = tmp_path / "team"
_ = _write_skill(root, "alpha", body="Use pdftotext.")
_ = _write_skill(root, "beta", body="Nothing to see.")
with _bound(tmp_path, root):
assert await skill_search.handler("pdftotext", skill="beta") == "(no matches)"
assert (await skill_search.handler("pdftotext", skill="nope")).startswith("error: unknown skill")
assert (await skill_search.handler("[")).startswith("error: invalid regex")
async def test_skill_search_without_skills(tmp_path: Path) -> None:
empty = tmp_path / "empty"
empty.mkdir()
with _bound(tmp_path, empty):
assert await skill_search.handler("anything") == "(no skills to search)"
async def test_skill_search_respects_max_matches(tmp_path: Path) -> None:
root = tmp_path / "team"
_ = _write_skill(root, "alpha", body="hit\nhit\nhit\nhit\n")
with _bound(tmp_path, root):
out = await skill_search.handler("hit", max_matches=2)
assert out.count("alpha/SKILL.md:") == 2
assert out.endswith("...[truncated at 2 matches]")
def test_format_skill_table_lists_roots_in_priority_order(tmp_path: Path) -> None:
root = tmp_path / "team"
_ = _write_skill(root, "acme")
store = SkillStore(
build_roots(SkillsConfig(paths=[str(root)], discover=[]), workspace=tmp_path, user_config_dir=tmp_path)
)
table = format_skill_table(store)
assert "skills: 1" in table
assert "acme [config-user]" in table
def test_skill_tool_names_cover_the_group() -> None:
expected = ("skill_list", "skill_read", "skill_search")
for name in expected:
assert name in SKILL_TOOL_NAMES
assert len(SKILL_TOOL_NAMES) == len(expected)