[Bug Fix] Replace skill bundles atomically instead of merging over them

src/ucode/skills_download.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import shutil
 from collections.abc import Callable
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from pathlib import Path
@@ -115,13 +116,23 @@ def should_download_skill(roots: list[Path], ref: SkillRef) -> bool:
 
 
 def write_skill(roots: list[Path], ref: SkillRef, files: dict[str, bytes]) -> None:
-    """Write ``ref``'s bundle (``{relpath: bytes}``) into every root.
+    """Write ``ref``'s bundle (``{relpath: bytes}``) into every root, replacing any existing copy.
 
-    The directory is named for the bundle, so it matches the ``name:`` an agent
-    reads from the written SKILL.md.
+    The directory is named for the bundle, so it matches the ``name:`` an agent reads from the
+    written SKILL.md. Each root is cleared before it is rewritten, so a file removed upstream does
+    not linger. A write interrupted partway leaves only that one directory incomplete, never a
+    stray copy elsewhere; the next write clears and rebuilds it, so a retry always converges on
+    the current bundle.
     """
+    if not files:
+        return
     for root in roots:
-        _write_bundle(root / ref.bundle_name, ref.bundle_name, files)
+        skill_dir = root / ref.bundle_name
+        if skill_dir.is_symlink():
+            skill_dir.unlink()
+        elif skill_dir.is_dir():
+            shutil.rmtree(skill_dir)
+        _write_bundle(skill_dir, ref.bundle_name, files)
 
 
 def _skill_installs(

tests/test_skills_download.py

@@ -106,6 +106,63 @@ def test_path_traversal_is_rejected(self, tmp_path):
         assert (roots[0] / "triage/SKILL.md").read_bytes() == b"ok"
         assert not (tmp_path / "escape.md").exists()
 
+    def test_replace_drops_files_removed_upstream(self, tmp_path):
+        roots = skill_dir_roots(str(tmp_path))
+        write_skill(roots, ref("triage"), {"SKILL.md": b"v1", "notes.md": b"old"})
+
+        write_skill(roots, ref("triage"), {"SKILL.md": b"v2"})
+
+        for root in roots:
+            assert (root / "triage/SKILL.md").read_bytes() == b"v2"
+            assert not (root / "triage/notes.md").exists()
+
+    def test_empty_bundle_keeps_existing_copy(self, tmp_path):
+        roots = skill_dir_roots(str(tmp_path))
+        write_skill(roots, ref("triage"), {"SKILL.md": b"v1"})
+
+        write_skill(roots, ref("triage"), {})
+
+        assert (roots[0] / "triage/SKILL.md").read_bytes() == b"v1"
+
+    def test_replaces_symlinked_bundle_without_touching_its_target(self, tmp_path):
+        roots = skill_dir_roots(str(tmp_path))
+        target = tmp_path / "real-skill"
+        target.mkdir()
+        (target / "keep.md").write_bytes(b"authored")
+        roots[0].mkdir(parents=True)
+        (roots[0] / "triage").symlink_to(target)
+
+        write_skill(roots, ref("triage"), {"SKILL.md": b"fresh"})
+
+        assert not (roots[0] / "triage").is_symlink()
+        assert (roots[0] / "triage/SKILL.md").read_bytes() == b"fresh"
+        assert (target / "keep.md").exists()
+
+    def test_leaves_only_the_bundle_dir(self, tmp_path):
+        roots = skill_dir_roots(str(tmp_path))
+
+        write_skill(roots, ref("triage"), {"SKILL.md": b"v1"})
+        write_skill(roots, ref("triage"), {"SKILL.md": b"v2"})
+
+        for root in roots:
+            assert [p.name for p in root.iterdir()] == ["triage"]
+
+    def test_recovers_from_interrupted_previous_write(self, tmp_path):
+        roots = skill_dir_roots(str(tmp_path))
+        for root in roots:
+            partial = root / "triage"
+            (partial / "scripts").mkdir(parents=True)
+            (partial / "stale.py").write_bytes(b"garbage")
+            (partial / "scripts/old.py").write_bytes(b"garbage")
+
+        write_skill(roots, ref("triage"), {"SKILL.md": b"good", "scripts/run.py": b"print(1)"})
+
+        for root in roots:
+            assert (root / "triage/SKILL.md").read_bytes() == b"good"
+            assert (root / "triage/scripts/run.py").read_bytes() == b"print(1)"
+            assert not (root / "triage/stale.py").exists()
+            assert not (root / "triage/scripts/old.py").exists()
+
 
 class TestFetchBundles:
     def test_empty_leaves_returns_empty_without_pool(self):