From c7441d4a430e4aee65890032c4f4ef32deac8d22 Mon Sep 17 00:00:00 2001 From: worldmozara Date: Tue, 22 Sep 2026 22:46:21 +0800 Subject: [PATCH] tools/skills: create and modify skills behind a grant MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit skill_create writes a SKILL.md (name, description, body) plus whatever bundled files the model lists, and skill_edit changes one file of an existing skill either by literal replacement, by appending, or by rewriting it. Both resolve every path inside the skill directory first, so a bad argument is answered before anything is written and before the human is asked anything; a rewritten SKILL.md is re-parsed so a skill that lost its description (or its frontmatter) says so instead of failing quietly. The store is reloaded after a write, which makes a new skill visible to the very next skill_list. Writing is a host policy, not a directory-access grant: the denylist wins first, then a session grant, then [skills].allow_write, then the host's confirm hook — asked once per skill directory, so one answer covers its files. A host without a hook denies, which keeps the library default safe; the confirm can also answer 'session' to cover the rest of the chat. --- src/plyngent/cli/display.py | 39 ++++ src/plyngent/skills/__init__.py | 3 + src/plyngent/skills/discovery.py | 27 ++- src/plyngent/tools/context.py | 3 + src/plyngent/tools/skills.py | 351 ++++++++++++++++++++++++++++++- tests/test_cli/test_display.py | 35 +++ tests/test_tools/test_skills.py | 221 ++++++++++++++++++- 7 files changed, 663 insertions(+), 16 deletions(-) diff --git a/src/plyngent/cli/display.py b/src/plyngent/cli/display.py index 9d24aed..a9ce0a5 100644 --- a/src/plyngent/cli/display.py +++ b/src/plyngent/cli/display.py @@ -754,6 +754,43 @@ def _skill_search_detail(_args_json: str, result: str) -> str: return _pretty_segments((f"({len(matches)} {match_unit} in {len(skills)} {skill_unit})", None)) +def _skill_create_prefix(args_json: str) -> str: + name = _json_str_arg(args_json, "name") or "?" + return _pretty_prefix(f"* Skill Create '{name}' ") + + +def _skill_create_detail(_args_json: str, result: str) -> str: + """``skill_create`` outcome: created (with any notes), or the failure.""" + if result.startswith("error:"): + return _pretty_segments((f"({result})", "red")) + detail = _pretty_segments(("(created)", "green")) + notes = result.partition("\nnotes: ")[2] + if notes: + detail += _pretty_segments((f" — {notes}", "yellow")) + return detail + + +def _skill_edit_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 Edit {subject} ") + + +def _skill_edit_detail(_args_json: str, result: str) -> str: + """``skill_edit`` outcome: the change made, or the failure.""" + if result.startswith("error:"): + return _pretty_segments((f"({result})", "red")) + first = result.splitlines()[0] if result else "" + match = re.search(r"\((.+)\)$", first) + text = match.group(1) if match is not None else "updated" + detail = _pretty_segments((f"({text})", "green")) + notes = result.partition("\nnotes: ")[2] + if notes: + detail += _pretty_segments((f" — {notes}", "yellow")) + return detail + + # 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), @@ -781,6 +818,8 @@ _PRETTY_LINES: dict[str, PrettyLine] = { "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), + "skill_create": PrettyLine(prefix=_skill_create_prefix, detail=_skill_create_detail), + "skill_edit": PrettyLine(prefix=_skill_edit_prefix, detail=_skill_edit_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), diff --git a/src/plyngent/skills/__init__.py b/src/plyngent/skills/__init__.py index e173f4d..b9b54f6 100644 --- a/src/plyngent/skills/__init__.py +++ b/src/plyngent/skills/__init__.py @@ -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_inside as resolve_inside 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 @@ -18,5 +19,7 @@ __all__ = [ "SkillStore", "build_roots", "parse_frontmatter", + "resolve_inside", + "resolve_skill_file", "validate_skill_name", ] diff --git a/src/plyngent/skills/discovery.py b/src/plyngent/skills/discovery.py index c7b465b..a5999a5 100644 --- a/src/plyngent/skills/discovery.py +++ b/src/plyngent/skills/discovery.py @@ -136,26 +136,35 @@ 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. +def resolve_inside(base: Path, relative: str) -> Path | None: + """Resolve *relative* under *base*; ``None`` when it is absolute or escapes. - 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. + Symlinks are followed by the resolve, so a link pointing outside *base* is + rejected instead of becoming a way around the boundary. """ - token = relative.strip() or SKILL_FILE + token = relative.strip() + if not token: + return None candidate = Path(token) if candidate.is_absolute(): return None try: - base = skill.path.resolve() - target = (base / candidate).resolve() - _ = target.relative_to(base) + root = base.resolve() + target = (root / candidate).resolve() + _ = target.relative_to(root) except OSError, ValueError: return None return target +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``. + """ + return resolve_inside(skill.path, relative.strip() or SKILL_FILE) + + class SkillStore: """Skills discovered under *roots*, highest-priority root first.""" diff --git a/src/plyngent/tools/context.py b/src/plyngent/tools/context.py index dd010ff..05e67d1 100644 --- a/src/plyngent/tools/context.py +++ b/src/plyngent/tools/context.py @@ -86,6 +86,9 @@ class SessionState: # Directory-access grants for this session (resolved path → mode). Config / # process-scoped grants live on WorkspacePolicy; see plyngent.tools.access. access_grants: dict[Path, AccessMode] = field(default_factory=dict) + # Skill directories the human allowed this chat to write (resolved paths); + # the standing grant lives in [skills].allow_write (tools/skills.py). + skill_write_roots: set[Path] = field(default_factory=set) def has_grant(self, tool_name: str) -> bool: return bool(self.grants.get(tool_name)) diff --git a/src/plyngent/tools/skills.py b/src/plyngent/tools/skills.py index 8172253..308c36e 100644 --- a/src/plyngent/tools/skills.py +++ b/src/plyngent/tools/skills.py @@ -13,15 +13,35 @@ directory-access path of ``request_directory_access``. from __future__ import annotations +from dataclasses import dataclass +from typing import TYPE_CHECKING, cast + 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.skills import ( + SKILL_FILE, + FrontmatterError, + Skill, + SkillStore, + parse_frontmatter, + resolve_inside, + resolve_skill_file, + validate_skill_name, +) 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 +from plyngent.tools.workspace import DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS + +if TYPE_CHECKING: + from collections.abc import Callable + from pathlib import Path + + from plyngent.tools.context import InstanceState, SessionState # Keeps a listing readable when a third-party root holds a pile of skills. _MAX_LISTED_ISSUES = 8 +_SKILL_WRITE_POLICY_KEY = "skill_write_policy" def get_skill_store() -> SkillStore | None: @@ -206,10 +226,339 @@ async def skill_search( return body +# -- writing (needs a grant) ------------------------------------------------- + + +@dataclass(frozen=True, slots=True) +class SkillWriteDecision: + """A human's answer to a skill write prompt: allow, and for how long.""" + + allow: bool + # True keeps the grant for the rest of this chat session (the standing + # alternative is ``[skills].allow_write`` in the config). + session: bool = False + + +type SkillWriteConfirmHook = Callable[[Path, str, str, float], SkillWriteDecision | bool | None] + + +@dataclass(frozen=True, slots=True) +class SkillWritePolicy: + """How this host authorizes writes to skill directories. + + ``allow_all`` is the standing grant (``[skills].allow_write``). Otherwise the + host's ``confirm`` hook is asked per skill directory; a host that installs no + hook denies, which keeps the library default safe. + """ + + allow_all: bool = False + confirm: SkillWriteConfirmHook | None = None + timeout: float = DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS + + +def set_skill_write_policy(policy: SkillWritePolicy | None, *, instance: InstanceState | None = None) -> None: + """Install (or clear) the host's skill-write policy on the bound instance.""" + from plyngent.tools.context import require_instance + + inst = instance if instance is not None else require_instance() + if policy is None: + _ = inst.extras.pop(_SKILL_WRITE_POLICY_KEY, None) + return + inst.extras[_SKILL_WRITE_POLICY_KEY] = policy + + +def get_skill_write_policy() -> SkillWritePolicy | None: + """The policy installed on the bound instance, if any.""" + from plyngent.tools.context import get_instance + + instance = get_instance() + return cast("SkillWritePolicy | None", instance.extras.get(_SKILL_WRITE_POLICY_KEY)) if instance else None + + +def _denied_by_policy(*paths: Path) -> str | None: + """Hard path denylist for a write target (never skippable by a grant).""" + from plyngent.tools.workspace import denylist_match + + for path in paths: + for candidate, directory in ((path, path.is_dir()), (path.parent, True)): + matched = denylist_match(candidate, directory=directory) + if matched is not None: + return f"error: path denied by policy (matched {matched!r}): {path}" + return None + + +def _prompt_write( + skill_dir: Path, + *, + action: str, + detail: str, + policy: SkillWritePolicy, + session: SessionState | None, +) -> str | None: + """Ask the host's hook and record a session grant when the human chose one.""" + hook = policy.confirm + if hook is None: + return None + try: + decision = hook(skill_dir, action, detail, policy.timeout) + except Exception as exc: # noqa: BLE001 — surface a failing prompt to the model + return f"error: skill write confirm failed: {exc}" + allowed = decision is True or (isinstance(decision, SkillWriteDecision) and decision.allow) + if not allowed: + return f"error: skill write to {skill_dir} denied (declined or timed out after {policy.timeout:g}s)" + if isinstance(decision, SkillWriteDecision) and decision.session and session is not None: + session.skill_write_roots.add(skill_dir) + return None + + +def _authorize_write(skill_dir: Path, target: Path, *, action: str, detail: str) -> str | None: + """``None`` when a write to *target* inside *skill_dir* is authorized. + + Grants are per **skill directory** — that is the unit the human is asked + about, and one answer covers every file of that skill. The denylist is + checked first and always wins; then a session grant, then + ``[skills].allow_write``, then the host's confirm hook. Without a hook the + write is denied rather than silently allowed. + """ + denied = _denied_by_policy(target, skill_dir) + if denied is not None: + return denied + from plyngent.tools.context import get_session + + session = get_session() + if session is not None and skill_dir in session.skill_write_roots: + return None + policy = get_skill_write_policy() + if policy is None or (policy.confirm is None and not policy.allow_all): + return ( + "error: writing skills needs a grant (no confirm hook is installed; denied); " + "set [skills].allow_write = true to allow skill writes" + ) + if policy.allow_all: + return None + return _prompt_write(skill_dir, action=action, detail=detail, policy=policy, session=session) + + +def _frontmatter_scalar(text: str) -> str: + """One-line, quoted-if-needed frontmatter value for generated SKILL.md files.""" + one_line = " ".join(text.split()) + if not one_line: + return '""' + if any(char in one_line for char in ":#{}[]\"'"): + return '"' + one_line.replace('"', "'") + '"' + return one_line + + +def _render_skill_md(name: str, description: str, body: str) -> str: + """Compose a minimal ``SKILL.md``: name, description, then the body.""" + front = f"---\nname: {name}\ndescription: {_frontmatter_scalar(description)}\n---\n\n" + return front + body.strip() + "\n" + + +def _skill_notes(skill_file: Path) -> str: + """Non-fatal notes after editing a ``SKILL.md`` (parse errors, missing fields).""" + try: + metadata, _body = parse_frontmatter(skill_file.read_text(encoding="utf-8", errors="replace")) + except (FrontmatterError, OSError) as exc: + return f"\nnotes: SKILL.md frontmatter not parsed ({exc}); it will use the directory name" + description = metadata.get("description") + if not isinstance(description, str) or not description.strip(): + return "\nnotes: SKILL.md has no frontmatter description; the first paragraph will be used" + return "" + + +def _write_file(target: Path, content: str) -> None: + """Write *content* under a skill, creating parent directories.""" + target.parent.mkdir(parents=True, exist_ok=True) + _ = target.write_text(content, encoding="utf-8") + + +def _create_target(store: SkillStore, name: str) -> tuple[Path, str] | str: + """Resolve the directory a new *name* goes into; error string when unusable.""" + problem = validate_skill_name(name) + if problem is not None: + return f"error: {problem}" + root = store.default_write_root() + if root is None: + return "error: no skill root is configured; set [skills].paths or [skills].discover" + token = name.strip() + target = root.path / token + if (target / SKILL_FILE).is_file(): + return f"error: skill {token!r} already exists at {target}; use skill_edit to change it" + if store.find(token) is not None: + return f"error: skill {token!r} is already visible from another root; edit it there instead" + return target, token + + +def _bundled_targets(target: Path, files: dict[str, str], *, name: str) -> dict[Path, str] | str: + """Resolve every bundled file path before anything is written.""" + resolved: dict[Path, str] = {} + for relative, content in files.items(): + path = resolve_inside(target, relative) + if path is None: + return f"error: bundled file {relative!r} is outside skill {name!r}" + resolved[path] = content + return resolved + + +@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE) +async def skill_create( + name: str, + description: str, + body: str, + *, + files: dict[str, str] | None = None, +) -> str: + """Create a new skill: a directory with a ``SKILL.md`` and optional files. + + ``name`` is the skill name (lowercase letters, digits, hyphens); the skill + is written under the default write root (our own user skills directory + unless ``[skills].paths`` puts an explicit one first). ``body`` is the + instruction text, and ``files`` maps paths inside the skill to their content + (scripts, references, templates). Writing needs a grant: the human is asked + once per skill, or the whole feature is opened with ``[skills].allow_write``. + """ + store = _store_or_error() + if isinstance(store, str): + return store + resolution = _create_target(store, name) + if isinstance(resolution, str): + return resolution + target, token = resolution + bundled = _bundled_targets(target, files or {}, name=token) + if isinstance(bundled, str): + return bundled + + detail = f"create skill {token!r} at {target} (SKILL.md" + detail += f" + {len(bundled)} file(s)" if bundled else "" + detail += ")" + denial = _authorize_write(target, target, action="create", detail=detail) + if denial is not None: + return denial + + try: + _write_file(target / SKILL_FILE, _render_skill_md(token, description, body)) + for path, content in bundled.items(): + _write_file(path, content) + except OSError as exc: + return f"error: failed to create skill {token!r}: {exc}" + store.reload() + return f"created skill {token!r} at {target}{_skill_notes(target / SKILL_FILE)}" + + +def _edit_target(store: SkillStore, name: str, file: str) -> tuple[Skill, Path, str] | str: + """Look up the skill and the file to edit inside it.""" + 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}" + return skill, target, file.strip() or SKILL_FILE + + +def _edit_mode_error(*, old_string: str, new_string: str, append: str) -> str | None: + """Reject an edit request that names no form, or two at once.""" + if not (old_string or new_string or append): + return "error: nothing to write; pass old_string + new_string, append, or new_string alone" + if old_string and append: + return "error: use either old_string + new_string or append, not both" + return None + + +def _apply_text_edit( + existing: str | None, + *, + relative: str, + skill_name: str, + old_string: str, + new_string: str, + append: str, + replace_all: bool, +) -> tuple[str, str] | str: + """Compute the new file text; ``(text, detail)`` or an error string.""" + mode_error = _edit_mode_error(old_string=old_string, new_string=new_string, append=append) + if mode_error is not None: + return mode_error + if old_string: + text = existing or "" + found = text.count(old_string) # non-overlapping, like str.replace + if found == 0: + return f"error: old_string not found in {relative} of skill {skill_name!r}" + limit = found if replace_all else 1 + detail = f"{relative}: replaced {limit} of {found} occurrence(s)" + return text.replace(old_string, new_string, limit), detail + if append: + if existing is None: + return f"error: no such file in skill {skill_name!r}: {relative}" + separator = "" if not existing or existing.endswith("\n") else "\n" + detail = f"{relative}: appended {len(append)} characters" + return f"{existing}{separator}{append}", detail + return new_string, f"{relative}: rewrote {len(new_string)} characters" + + +@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE) +async def skill_edit( + name: str, + *, + file: str = SKILL_FILE, + old_string: str = "", + new_string: str = "", + append: str = "", + replace_all: bool = False, +) -> str: + """Change one file of an existing skill (``SKILL.md`` by default). + + Three mutually exclusive forms: + + - ``old_string`` + ``new_string``: literal replacement of the first match, + or of every match with ``replace_all``; + - ``append``: add text at the end of the file; + - ``new_string`` alone: rewrite the whole file (a missing file is created, + parent directories included). + + Writing needs a grant for that skill (asked once, or standing via + ``[skills].allow_write``). + """ + store = _store_or_error() + if isinstance(store, str): + return store + resolution = _edit_target(store, name, file) + if isinstance(resolution, str): + return resolution + skill, target, relative = resolution + existing = target.read_text(encoding="utf-8", errors="replace") if target.is_file() else None + outcome = _apply_text_edit( + existing, + relative=relative, + skill_name=skill.name, + old_string=old_string, + new_string=new_string, + append=append, + replace_all=replace_all, + ) + if isinstance(outcome, str): + return outcome + updated, detail = outcome + + denial = _authorize_write(skill.path, target, action="update", detail=f"{skill.name}: {detail}") + if denial is not None: + return denial + try: + _write_file(target, updated) + except OSError as exc: + return f"error: failed to write {relative} of skill {skill.name!r}: {exc}" + store.reload() + notes = _skill_notes(target) if target.name == SKILL_FILE else "" + return f"updated skill {skill.name!r} ({detail}){notes}" + + SKILL_TOOLS = [ skill_list, skill_read, skill_search, + skill_create, + skill_edit, ] # Fed to the catalog selection filter: ``[skills].enabled = false`` drops these diff --git a/tests/test_cli/test_display.py b/tests/test_cli/test_display.py index d3c01ff..10dfcfb 100644 --- a/tests/test_cli/test_display.py +++ b/tests/test_cli/test_display.py @@ -868,6 +868,41 @@ async def test_pretty_skill_tools(capsys: pytest.CaptureFixture[str]) -> None: assert "* Skill Search 'nope' (no matches)" in out +async def test_pretty_skill_create_and_edit(capsys: pytest.CaptureFixture[str]) -> None: + await render_events( + _aiter( + [ + _pretty_call("skill_create", '{"name": "acme", "description": "d", "body": "b"}'), + _result("created skill 'acme' at /skills/acme\nnotes: SKILL.md has no frontmatter description"), + ] + ) + ) + out = capsys.readouterr().out + assert "* Skill Create 'acme' (created) — SKILL.md has no frontmatter description" in out + + await render_events( + _aiter( + [ + _pretty_call("skill_edit", '{"name": "acme"}'), + _result("updated skill 'acme' (SKILL.md: replaced 1 of 1 occurrence(s))"), + ] + ) + ) + out = capsys.readouterr().out + assert "* Skill Edit 'acme' (SKILL.md: replaced 1 of 1 occurrence(s))" in out + + await render_events( + _aiter( + [ + _pretty_call("skill_edit", '{"name": "acme", "file": "scripts/run.sh"}'), + _result("error: skill write to /skills/acme denied (declined or timed out after 30s)"), + ] + ) + ) + out = capsys.readouterr().out + assert "* Skill Edit 'acme' scripts/run.sh (error: skill write to /skills/acme denied" in out + + async def test_pretty_prefix_shown_while_call_runs( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], diff --git a/tests/test_tools/test_skills.py b/tests/test_tools/test_skills.py index ab64427..0266eca 100644 --- a/tests/test_tools/test_skills.py +++ b/tests/test_tools/test_skills.py @@ -5,9 +5,20 @@ 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 +from plyngent.tools.context import InstanceState, SessionState, bind_instance, bind_session +from plyngent.tools.skills import ( + SKILL_TOOL_NAMES, + SkillWriteDecision, + SkillWritePolicy, + format_skill_table, + set_skill_write_policy, + skill_create, + skill_edit, + skill_list, + skill_read, + skill_search, +) +from plyngent.tools.workspace import set_path_denylist, set_static_read_roots, set_workspace_root if TYPE_CHECKING: from collections.abc import Generator @@ -26,7 +37,13 @@ def _write_skill(root: Path, name: str, *, description: str | None = "A test ski @contextmanager -def _bound(tmp_path: Path, *roots: Path, skills: SkillStore | None = None) -> Generator[InstanceState]: +def _bound( + tmp_path: Path, + *roots: Path, + skills: SkillStore | None = None, + session: bool = True, + policy: SkillWritePolicy | 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) @@ -40,12 +57,30 @@ def _bound(tmp_path: Path, *roots: Path, skills: SkillStore | None = None) -> Ge ) ) instance = InstanceState(workspace_root=workspace, skills=store) - with bind_instance(instance): + session_state = SessionState(session_id=1) if session else None + with bind_instance(instance), bind_session(session_state): _ = set_workspace_root(workspace) _ = set_static_read_roots(roots) + if policy is not None: + set_skill_write_policy(policy, instance=instance) yield instance +def _write_root(tmp_path: Path) -> Path: + """An empty user-level skills directory to create skills in.""" + root = tmp_path / "ours" + root.mkdir(parents=True, exist_ok=True) + return root + + +def _allow(*_args: object) -> SkillWriteDecision: + return SkillWriteDecision(allow=True) + + +def _deny(*_args: object) -> bool: + return False + + 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 @@ -188,7 +223,181 @@ def test_format_skill_table_lists_roots_in_priority_order(tmp_path: Path) -> Non def test_skill_tool_names_cover_the_group() -> None: - expected = ("skill_list", "skill_read", "skill_search") + expected = ("skill_list", "skill_read", "skill_search", "skill_create", "skill_edit") for name in expected: assert name in SKILL_TOOL_NAMES assert len(SKILL_TOOL_NAMES) == len(expected) + + +async def test_skill_create_writes_skill_md_and_bundled_files(tmp_path: Path) -> None: + root = _write_root(tmp_path) + with _bound(tmp_path, root, policy=SkillWritePolicy(allow_all=True)): + out = await skill_create.handler( + "acme-deploy", + "Deploy the acme stack", + "# Steps\n\n1. Build.\n", + files={"scripts/deploy.sh": "#!/bin/sh\necho deploy\n"}, + ) + assert out.startswith("created skill 'acme-deploy' at ") + assert (root / "acme-deploy" / SKILL_FILE).is_file() + assert (root / "acme-deploy" / "scripts" / "deploy.sh").read_text(encoding="utf-8").startswith("#!/bin/sh") + # The new skill is visible right away, without a restart. + assert "acme-deploy [config-user]" in await skill_list.handler() + read = await skill_read.handler("acme-deploy") + assert "description: Deploy the acme stack" in read + assert "1. Build." in read + + +async def test_skill_create_quotes_a_frontmatter_description_that_needs_it(tmp_path: Path) -> None: + root = _write_root(tmp_path) + with _bound(tmp_path, root, policy=SkillWritePolicy(allow_all=True)): + _ = await skill_create.handler("quoted", "Use: colons, {braces}", "Body.") + skill = SkillStore( + build_roots(SkillsConfig(paths=[str(root)], discover=[]), workspace=tmp_path, user_config_dir=tmp_path) + ).skills[0] + assert skill.description == "Use: colons, {braces}" + + +async def test_skill_create_rejects_a_bad_name_and_a_taken_one(tmp_path: Path) -> None: + root = _write_root(tmp_path) + _ = _write_skill(root, "taken") + with _bound(tmp_path, root, policy=SkillWritePolicy(allow_all=True)): + assert (await skill_create.handler("Not-A-Name", "d", "b")).startswith("error: skill name") + assert "already exists" in await skill_create.handler("taken", "d", "b") + assert (await skill_create.handler("ok-name", "d", "b")).startswith("created skill") + + +async def test_skill_create_rejects_a_bundled_file_outside_the_skill(tmp_path: Path) -> None: + root = _write_root(tmp_path) + with _bound(tmp_path, root, policy=SkillWritePolicy(allow_all=True)): + out = await skill_create.handler("acme", "d", "b", files={"../escape.sh": "x"}) + assert out == "error: bundled file '../escape.sh' is outside skill 'acme'" + assert not (tmp_path / "escape.sh").exists() + + +async def test_skill_write_needs_a_grant(tmp_path: Path) -> None: + """Without a hook the write is denied, and the message says how to open it.""" + root = _write_root(tmp_path) + with _bound(tmp_path, root): + out = await skill_create.handler("acme", "d", "b") + assert out == ( + "error: writing skills needs a grant (no confirm hook is installed; denied); " + "set [skills].allow_write = true to allow skill writes" + ) + assert not (root / "acme").exists() + + +async def test_skill_write_prompt_denial_is_reported(tmp_path: Path) -> None: + root = _write_root(tmp_path) + _ = _write_skill(root, "acme", body="Before.") + with _bound(tmp_path, root, policy=SkillWritePolicy(confirm=_deny, timeout=3.0)): + out = await skill_edit.handler("acme", old_string="Before.", new_string="After.") + assert out.startswith("error: skill write to ") + assert out.endswith("denied (declined or timed out after 3s)") + assert "After." not in (root / "acme" / SKILL_FILE).read_text(encoding="utf-8") + + +async def test_skill_write_confirm_sees_the_action_and_the_skill(tmp_path: Path) -> None: + root = _write_root(tmp_path) + seen: list[tuple[Path, str, str, float]] = [] + + def hook(target: Path, action: str, detail: str, timeout: float) -> bool: + seen.append((target, action, detail, timeout)) + return True + + with _bound(tmp_path, root, policy=SkillWritePolicy(confirm=hook, timeout=5.0)): + _ = await skill_create.handler("acme", "d", "b") + assert seen == [(root / "acme", "create", f"create skill 'acme' at {root / 'acme'} (SKILL.md)", 5.0)] + + +async def test_skill_write_session_grant_is_asked_once(tmp_path: Path) -> None: + """A 'session' answer covers the same skill for the rest of the chat.""" + root = _write_root(tmp_path) + calls: list[Path] = [] + + def hook(target: Path, action: str, detail: str, timeout: float) -> SkillWriteDecision: + calls.append(target) + return SkillWriteDecision(allow=True, session=True) + + with _bound(tmp_path, root, policy=SkillWritePolicy(confirm=hook)): + _ = await skill_create.handler("acme", "d", "b") + _ = await skill_edit.handler("acme", append="\nMore.\n") + _ = await skill_edit.handler("acme", file="notes.md", new_string="note\n") + # One prompt for the skill directory, covering SKILL.md and later files. + assert calls == [root / "acme"] + assert (root / "acme" / "notes.md").read_text(encoding="utf-8") == "note\n" + + +async def test_skill_write_denylist_wins_over_a_grant(tmp_path: Path) -> None: + root = _write_root(tmp_path) + with _bound(tmp_path, root, policy=SkillWritePolicy(allow_all=True)): + set_path_denylist(["acme"]) + try: + out = await skill_create.handler("acme", "d", "b") + finally: + set_path_denylist(()) + assert out.startswith("error: path denied by policy (matched 'acme')") + assert not (root / "acme").exists() + + +async def test_skill_edit_forms(tmp_path: Path) -> None: + root = _write_root(tmp_path) + _ = _write_skill(root, "acme", body="Line one.\nLine two.") + with _bound(tmp_path, root, policy=SkillWritePolicy(allow_all=True)): + out = await skill_edit.handler("acme", old_string="Line two.", new_string="Line two, edited.") + assert out.startswith("updated skill 'acme' (SKILL.md: replaced 1 of 1 occurrence(s))") + + out = await skill_edit.handler("acme", append="\nAppended tail.") + assert "appended 15 characters" in out + + out = await skill_edit.handler("acme", file="scripts/run.sh", new_string="#!/bin/sh\n") + assert "rewrote 10 characters" in out + assert (root / "acme" / "scripts" / "run.sh").is_file() + text = (root / "acme" / SKILL_FILE).read_text(encoding="utf-8") + assert "Line two, edited." in text + assert text.endswith("Appended tail.") + + +async def test_skill_edit_replace_all_and_missing_match(tmp_path: Path) -> None: + root = _write_root(tmp_path) + directory = _write_skill(root, "acme", body="hit\nhit\nhit") + with _bound(tmp_path, root, policy=SkillWritePolicy(allow_all=True)): + out = await skill_edit.handler("acme", old_string="hit", new_string="done", replace_all=True) + assert "replaced 3 of 3 occurrence(s)" in out + missing = await skill_edit.handler("acme", old_string="nope", new_string="x") + assert missing == "error: old_string not found in SKILL.md of skill 'acme'" + assert (directory / SKILL_FILE).read_text(encoding="utf-8").count("done") == 3 + + +async def test_skill_edit_rejects_conflicting_forms(tmp_path: Path) -> None: + root = _write_root(tmp_path) + _ = _write_skill(root, "acme") + with _bound(tmp_path, root, policy=SkillWritePolicy(allow_all=True)): + assert (await skill_edit.handler("acme")).startswith("error: nothing to write") + assert (await skill_edit.handler("acme", old_string="a", new_string="b", append="c")).startswith( + "error: use either" + ) + assert (await skill_edit.handler("acme", file="nope.md", append="x")) == ( + "error: no such file in skill 'acme': nope.md" + ) + assert (await skill_edit.handler("acme", file="../x.md", new_string="x")).startswith("error: file") + + +async def test_skill_edit_notes_a_rewritten_frontmatter(tmp_path: Path) -> None: + """A SKILL.md that loses its description (or breaks) is reported, not repaired.""" + root = _write_root(tmp_path) + _ = _write_skill(root, "acme") + with _bound(tmp_path, root, policy=SkillWritePolicy(allow_all=True)): + dropped = await skill_edit.handler("acme", new_string="Only a body, no frontmatter.\n") + broken = await skill_edit.handler("acme", new_string="---\nname: acme\nbroken line\n---\n\nBody.\n") + assert dropped.endswith("notes: SKILL.md has no frontmatter description; the first paragraph will be used") + assert "notes: SKILL.md frontmatter not parsed" in broken + + +async def test_skill_tools_reject_writes_when_no_session_is_bound(tmp_path: Path) -> None: + """A host without session state still denies instead of crashing on the grant.""" + root = _write_root(tmp_path) + _ = _write_skill(root, "acme", body="Before.") + with _bound(tmp_path, root, session=False, policy=SkillWritePolicy(confirm=_allow)): + out = await skill_edit.handler("acme", old_string="Before.", new_string="After.") + assert out.startswith("updated skill 'acme'")