fix(skills): skill_manage follows directory symlinks; rollback and ownership stay on the link

agent/skill_utils.py

@@ -610,17 +610,30 @@ def _resolve_for_skill_ownership(path) -> Path:
         return path_obj.expanduser().absolute()
 
 
+def _lexical_for_skill_ownership(path) -> Path:
+    """``path`` absolute and ``..``-normalised WITHOUT following symlinks: where a skill is
+    installed, as opposed to where its link points."""
+    path_obj = path if isinstance(path, Path) else Path(str(path))
+    return Path(os.path.abspath(os.path.expanduser(str(path_obj))))
+
+
 def is_external_skill_path(path) -> bool:
     """True when ``path`` lives under an external or trusted project skills dir.
     Those are externally owned: autonomous lifecycle maintenance treats them as
-    read-only (user-directed tool calls may still edit them)."""
-    candidate = _resolve_for_skill_ownership(path)
+    read-only (user-directed tool calls may still edit them).
+
+    Ownership follows where the skill is INSTALLED: a directory symlink under an external
+    root (``<root>/<skill> -> ../_src/<skill>``) is external even though it resolves
+    elsewhere, so the lexical path is checked before the resolved one (#54195)."""
+    lexical = _lexical_for_skill_ownership(path)
+    resolved = _resolve_for_skill_ownership(path)
     roots: List[Path] = list(get_external_skills_dirs())
     try:
         roots.extend(get_project_skills_dirs())
     except Exception:
         pass
-    return any(candidate.is_relative_to(_resolve_for_skill_ownership(root)) for root in roots)
+    return any(lexical.is_relative_to(_lexical_for_skill_ownership(root))
+               or resolved.is_relative_to(_resolve_for_skill_ownership(root)) for root in roots)
 
 
 def _hermes_metadata(frontmatter: Dict[str, Any]) -> Dict[str, Any]:

tests/tools/test_skill_manager_tool.py

@@ -824,6 +824,133 @@ def test_background_review_fails_closed_when_ownership_lookup_errors(self, tmp_p
             tmp_path / "manual-skill" / "SKILL.md"
         ).read_text(encoding="utf-8")
 
+
+def _link_external_skill(tmp_path: Path, name: str = "linked-skill"):
+    """A shared skill installed into the external root as a directory symlink — the
+    ``<external>/<skill> -> <elsewhere>/_src/<skill>`` layout a shared-skills tree uses.
+    Returns ``(local, external, source)``; skips where directory symlinks are unavailable."""
+    local, external = tmp_path / "local", tmp_path / "external"
+    source = tmp_path / "shared" / "_src" / name
+    local.mkdir(); external.mkdir(); source.mkdir(parents=True)
+    (source / "SKILL.md").write_text(
+        f"---\nname: {name}\ndescription: A shared skill.\n---\n\n"
+        "# Shared\n\nBody with OLD_MARKER here.\n", encoding="utf-8")
+    try:
+        (external / name).symlink_to(source, target_is_directory=True)
+    except (OSError, NotImplementedError):
+        pytest.skip("Directory symlinks are unavailable on this platform")
+    return local, external, source
+
+
+def _batch(ops):
+    return json.loads(skill_manage(action="", name="", operations=ops))
+
+
+class TestSymlinkedExternalSkill:
+    """A skill reached through a directory symlink inside skills.external_dirs (the shared-skills
+    layout ``<root>/<skill> -> ../_src/<skill>``). skill_view already follows such links
+    (``iter_skill_index_files`` walks with followlinks=True); skill_manage must resolve the same
+    skill and mutate it in place, through the public operations[] call. Regression for #54195.
+    """
+
+    def test_batch_patch_follows_directory_symlink_in_place(self, tmp_path):
+        local, external, source = _link_external_skill(tmp_path)
+        with _two_roots(local, external):
+            result = _batch([{"name": "linked-skill", "action": "patch",
+                              "old_string": "OLD_MARKER", "new_string": "NEW_MARKER"}])
+        assert result["success"] is True, result
+        assert (external / "linked-skill").is_symlink()          # alias untouched
+        assert "NEW_MARKER" in (source / "SKILL.md").read_text(encoding="utf-8")
+        assert not (local / "linked-skill").exists()             # no duplicate in local
+
+    def test_failed_single_op_batch_keeps_symlink_and_source(self, tmp_path):
+        """Rollback must land on the link TARGET: restoring onto the alias renames the link
+        aside and copies the snapshot into a real directory, silently detaching the skill
+        from its shared source (and leaving .rollback-broken debris)."""
+        local, external, source = _link_external_skill(tmp_path)
+        with _two_roots(local, external):
+            result = _batch([{"name": "linked-skill", "action": "patch",
+                              "old_string": "NOT_IN_FILE", "new_string": "x"}])
+        assert result["success"] is False
+        alias = external / "linked-skill"
+        assert alias.is_symlink() and alias.resolve() == source.resolve()
+        assert "OLD_MARKER" in (source / "SKILL.md").read_text(encoding="utf-8")
+        assert sorted(p.name for p in external.iterdir()) == ["linked-skill"]
+
+    def test_failed_multi_op_batch_rolls_back_shared_source(self, tmp_path):
+        """The first op lands in the shared source through the link; a later failure must
+        undo it THERE, not in a fresh copy at the alias."""
+        local, external, source = _link_external_skill(tmp_path)
+        with _two_roots(local, external):
+            result = _batch([
+                {"name": "linked-skill", "action": "patch",
+                 "old_string": "OLD_MARKER", "new_string": "NEW_MARKER"},
+                {"name": "linked-skill", "action": "patch",
+                 "old_string": "NOT_IN_FILE", "new_string": "x"}])
+        assert result["success"] is False and result["failed_index"] == 1
+        alias = external / "linked-skill"
+        assert alias.is_symlink() and alias.resolve() == source.resolve()
+        content = (source / "SKILL.md").read_text(encoding="utf-8")
+        assert "OLD_MARKER" in content and "NEW_MARKER" not in content
+        assert sorted(p.name for p in external.iterdir()) == ["linked-skill"]
+
+    def test_background_review_refuses_linked_external_skill(self, tmp_path):
+        """Autonomous curation treats skills.external_dirs as read-only. Ownership follows where
+        the skill is INSTALLED (the alias under the external root), not where the link points:
+        resolving first lets a curator-managed record open the shared source to an autonomous
+        write. Everything else that could refuse is satisfied here, so only ownership stands
+        between the write and the source."""
+        from tools.skill_manager_guards import mark_background_review_skill_read
+        from tools.skill_provenance import (
+            BACKGROUND_REVIEW,
+            reset_current_write_origin,
+            set_current_write_origin,
+        )
+
+        local, external, source = _link_external_skill(tmp_path)
+        curated = {"linked-skill": {"created_by": "agent", "pinned": False}}
+        token = set_current_write_origin(BACKGROUND_REVIEW)
+        try:
+            with _two_roots(local, external), \
+                 patch("agent.skill_utils.get_external_skills_dirs", return_value=[external]), \
+                 patch("agent.skill_utils.get_project_skills_dirs", return_value=[]), \
+                 patch("tools.skill_usage.load_usage", return_value=curated), \
+                 patch("tools.skill_usage.get_record", side_effect=lambda n: curated.get(n, {})), \
+                 patch("tools.skill_usage.is_protected_builtin", return_value=False), \
+                 patch("tools.skill_usage.is_hub_installed", return_value=False), \
+                 patch("tools.skill_usage.is_bundled", return_value=False):
+                mark_background_review_skill_read(source / "SKILL.md")
+                result = _batch([{"name": "linked-skill", "action": "patch",
+                                  "old_string": "OLD_MARKER", "new_string": "NEW_MARKER"}])
+        finally:
+            reset_current_write_origin(token)
+        assert result["success"] is False, result
+        assert "external" in result["error"].lower()
+        assert "OLD_MARKER" in (source / "SKILL.md").read_text(encoding="utf-8")
+
+    def test_failed_create_batch_never_deletes_through_a_link(self, tmp_path):
+        """A batch-created skill is removed at its LEXICAL path on rollback. When an alias already
+        sat at that path (no SKILL.md behind it, so 'create' saw no skill and wrote through the
+        link), rollback must refuse the link — as rmtree does on a plain path — rather than
+        resolve it and delete the target's contents."""
+        local = tmp_path / "local"
+        target = tmp_path / "shared" / "_src" / "aliased"
+        local.mkdir(); target.mkdir(parents=True)
+        (target / "notes.txt").write_text("keep me", encoding="utf-8")
+        try:
+            (local / "aliased").symlink_to(target, target_is_directory=True)
+        except (OSError, NotImplementedError):
+            pytest.skip("Directory symlinks are unavailable on this platform")
+        with _skill_dir(local):
+            result = _batch([
+                {"name": "aliased", "action": "create", "content": VALID_SKILL_CONTENT},
+                {"name": "aliased", "action": "write_file",
+                 "file_path": "bad/nope.md", "file_content": "x"}])
+        assert result["success"] is False
+        assert (local / "aliased").is_symlink()
+        assert (target / "notes.txt").read_text(encoding="utf-8") == "keep me"
+
+
 class TestBackgroundOwnershipPolicyConsistency:
     """The autonomous write policy must not depend on its own side effects.
 

tools/skill_manager_batch.py

@@ -8,6 +8,7 @@
 import shutil
 import tempfile
 from pathlib import Path
+from typing import Optional
 
 logger = logging.getLogger("tools.skill_manager_tool")
 
@@ -57,12 +58,26 @@ def fail(i, msg):
     return names, None
 
 
+def _real_skill_dir(found) -> Optional[Path]:
+    """The directory snapshot/restore operate on. Discovery returns the LEXICAL path, which may
+    be (or sit under) a directory symlink — a shared skill installed as an alias. Rolling back on
+    the alias renames the link aside and copies the snapshot into a real directory at the alias,
+    silently detaching the skill from its shared source (which keeps any op that already landed).
+    Resolve so both the copy-aside and the restore happen on the link target."""
+    if not found:
+        return None
+    path = Path(found["path"])
+    try:
+        return path.resolve()
+    except OSError:
+        return path
+
+
 def _snapshot_skills(names, snap_root, find_skill):
     """Copy every touched skill aside. Returns (snapshots, None) or (None, error_text)."""
     snapshots = {}  # skill name -> (pre_dir or None, snapshot_dir or None)
     for nm in dict.fromkeys(names):  # ordered unique
-        pre = find_skill(nm)
-        pre_dir = Path(pre["path"]) if pre else None
+        pre_dir = _real_skill_dir(find_skill(nm))
         snap = snap_root / nm if pre_dir is not None and pre_dir.is_dir() else None
         if snap is not None:
             try:
@@ -103,7 +118,11 @@ def _rollback(snapshots, find_skill):
     for nm, (pre_dir, snap) in snapshots.items():
         try:
             post = find_skill(nm)
-            _restore_snapshot(pre_dir, snap, Path(post["path"]) if post else None)
+            # A batch-created skill (no snapshot) is removed at its LEXICAL path: if an alias
+            # already sat there, rmtree refuses the link (loud rollback failure, as on a plain
+            # create) instead of following it into a shared tree.
+            post_dir = (Path(post["path"]) if post else None) if snap is None else _real_skill_dir(post)
+            _restore_snapshot(pre_dir, snap, post_dir)
         except Exception as exc:  # noqa: BLE001
             notes.append(f"ROLLBACK FAILED for '{nm}' ({exc})"
                          + (f"; snapshot preserved at '{snap}'" if snap is not None else ""))

tools/skill_manager_tool.py

@@ -193,8 +193,11 @@ def _resolve_skill_dir(name: str, category: str = None) -> Path:
 
 
 def _iter_skill_dirs(root: Path):
-    from agent.skill_utils import is_excluded_skill_path
-    for skill_md in root.rglob("SKILL.md"):
+    """Skill dirs under ``root`` via the shared reader-side walker (``os.walk(followlinks=True)``),
+    so a skill installed as a directory symlink resolves here exactly as it does for skill_view;
+    ``Path.rglob`` never descends into a linked directory (#54195)."""
+    from agent.skill_utils import is_excluded_skill_path, iter_skill_index_files
+    for skill_md in iter_skill_index_files(root, "SKILL.md"):
         if not is_excluded_skill_path(skill_md):
             yield skill_md.parent