tools/skills: create and modify skills behind a grant

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

@@ -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),

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",
 ]

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:
+def resolve_inside(base: Path, relative: str) -> Path | None:
-    """Resolve *relative* inside *skill*; ``None`` when it escapes the directory.
+    """Resolve *relative* under *base*; ``None`` when it is absolute or escapes.
 
-    The default (empty *relative*) is the skill's own ``SKILL.md``. Symlinks are
+    Symlinks are followed by the resolve, so a link pointing outside *base* is
-    followed by the resolve, so a link pointing outside the skill is rejected
+    rejected instead of becoming a way around the boundary.
-    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()
+        root = base.resolve()
-        target = (base / candidate).resolve()
+        target = (root / candidate).resolve()
-        _ = target.relative_to(base)
+        _ = 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."""
 

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))

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

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],

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.context import InstanceState, SessionState, bind_instance, bind_session
-from plyngent.tools.skills import SKILL_TOOL_NAMES, format_skill_table, skill_list, skill_read, skill_search
+from plyngent.tools.skills import (
-from plyngent.tools.workspace import set_static_read_roots, set_workspace_root
+    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'")