fix(installer): preserve skill edits when installation rolls back

loopx/windows_install.py

@@ -382,14 +382,48 @@ def _snapshot_paths(
     return snapshots
 
 
-def _restore_paths(snapshots: Sequence[tuple[Path, Path | None]]) -> None:
-    for path, backup in reversed(snapshots):
-        _remove_path(path)
-        if backup is not None:
-            _copy_path(backup, path)
+def _restore_paths(
+    snapshots: Sequence[tuple[Path, Path | None]],
+    *,
+    skill_paths: Sequence[Path] = (),
+) -> list[Path]:
+    skill_entries = set(skill_paths)
+    retained: list[Path] = []
+    try:
+        for path, backup in reversed(snapshots):
+            # A skill may be absent after deletion but before its copy starts.
+            if path.is_symlink() or path.is_dir() or (
+                path in skill_entries and path.exists()
+            ):
+                # A failed installation may have left user edits in a skill.
+                # Keep that state outside the temporary snapshot cleanup; its
+                # contents cannot be distinguished from installer writes here.
+                recovery_dir = Path(
+                    tempfile.mkdtemp(
+                        prefix=f".{path.name}.install-recovery.",
+                        dir=path.parent,
+                    )
+                )
+                recovery = recovery_dir / path.name
+                path.replace(recovery)
+                retained.append(recovery)
+            else:
+                _remove_path(path)
+            if backup is not None:
+                _copy_path(backup, path)
+    except Exception as exc:
+        if retained:
+            raise RuntimeError(
+                f"{exc}; interrupted skill state retained at: "
+                + ", ".join(str(path) for path in retained)
+            ) from exc
+        raise
+    return retained
 
 
-def _install_skills(release_root: Path, skills_dir: Path, installed_at: str) -> list[str]:
+def _install_skills(
+    release_root: Path, skills_dir: Path, installed_at: str
+) -> list[str]:
     skills_dir.mkdir(parents=True, exist_ok=True)
     installed_ids: list[str] = []
     for skill_id in PACKAGED_HOST_SKILL_IDS:
@@ -490,8 +524,8 @@ def install_windows(
             "installed_at": installed_at,
         }
         mutation_paths = [launcher, pointer, launcher_pointer]
-        if install_skills:
-            mutation_paths.extend(_skill_mutation_paths(skills_dir))
+        skill_paths = _skill_mutation_paths(skills_dir) if install_skills else []
+        mutation_paths.extend(skill_paths)
         with tempfile.TemporaryDirectory(
             prefix=".install-rollback.", dir=install_root
         ) as rollback_dir:
@@ -510,8 +544,9 @@ def install_windows(
                 _atomic_json(launcher_pointer, pointer_payload)
             except Exception as exc:
                 rollback_error: Exception | None = None
+                retained: list[Path] = []
                 try:
-                    _restore_paths(snapshots)
+                    retained = _restore_paths(snapshots, skill_paths=skill_paths)
                 except Exception as restore_exc:  # pragma: no cover - filesystem failure
                     rollback_error = restore_exc
                 # Keep the candidate: physical data may already have upgraded.
@@ -521,6 +556,12 @@ def install_windows(
                         "Windows installation failed and rollback was incomplete: "
                         f"{rollback_error}"
                     ) from exc
+                if retained:
+                    raise RuntimeError(
+                        f"{exc}; filesystem snapshots restored. "
+                        "Interrupted skill state retained at: "
+                        + ", ".join(str(path) for path in retained)
+                    ) from exc
                 raise
 
     return {

tests/test_windows_install.py

@@ -5,6 +5,8 @@
 import shutil
 import subprocess
 import sys
+import tempfile
+from types import SimpleNamespace
 import venv
 from pathlib import Path
 from typing import Any
@@ -305,6 +307,266 @@ def run(command: list[str], **_kwargs: object) -> subprocess.CompletedProcess[st
         )
 
 
+@pytest.mark.parametrize("previously_installed", [False, True])
+def test_skill_rollback_keeps_interrupted_directory_after_snapshot_cleanup(
+    tmp_path: Path, previously_installed: bool
+) -> None:
+    skill = tmp_path / "skills" / "browser-skill"
+    skill.parent.mkdir()
+    if previously_installed:
+        skill.mkdir()
+        (skill / "note.txt").write_text("Previous installed version\n")
+    with tempfile.TemporaryDirectory(dir=tmp_path) as temporary:
+        snapshots = windows_install._snapshot_paths([skill], Path(temporary))
+        skill.mkdir(exist_ok=True)
+        (skill / "note.txt").write_text("Customization made during installation\n")
+        retained = windows_install._restore_paths(snapshots)
+
+    if previously_installed:
+        assert (skill / "note.txt").read_text() == "Previous installed version\n"
+    else:
+        assert not skill.exists()
+    assert any(
+        path.read_text() == "Customization made during installation\n"
+        for path in tmp_path.rglob("note.txt")
+    ), "Rolling back must not delete the only copy of a concurrent customization"
+    assert len(retained) == 1
+    assert retained[0].parent.parent == skill.parent
+    assert (
+        retained[0] / "note.txt"
+    ).read_text() == "Customization made during installation\n"
+
+
+@pytest.mark.parametrize("relative_link", [False, True])
+def test_skill_rollback_keeps_replacement_symlink_without_mutating_its_target(
+    tmp_path: Path,
+    relative_link: bool,
+) -> None:
+    previous = tmp_path / "previous"
+    custom = tmp_path / "custom"
+    for target in (previous, custom):
+        target.mkdir()
+        (target / "note.txt").write_text(target.name)
+    skill = tmp_path / "browser-skill"
+    try:
+        skill.symlink_to(
+            previous.name if relative_link else previous, target_is_directory=True
+        )
+    except OSError:
+        pytest.skip("The host does not permit directory symlinks")
+    with tempfile.TemporaryDirectory(dir=tmp_path) as temporary:
+        snapshots = windows_install._snapshot_paths([skill], Path(temporary))
+        skill.unlink()
+        skill.symlink_to(
+            custom.name if relative_link else custom, target_is_directory=True
+        )
+        interrupted_target = os.readlink(skill)
+        retained = windows_install._restore_paths(snapshots)
+
+    assert skill.is_symlink() and skill.resolve() == previous
+    assert any(
+        path.is_symlink() and os.readlink(path) == interrupted_target
+        for path in tmp_path.rglob("*")
+    ), "The replacement link must remain recoverable after rollback"
+    assert len(retained) == 1 and retained[0].is_symlink()
+    assert os.readlink(retained[0]) == interrupted_target
+    # Archived relative links retain their literal target; returning the link
+    # to its original location restores that target's meaning without editing it.
+    skill.unlink()
+    retained[0].replace(skill)
+    assert skill.resolve() == custom
+    assert (previous / "note.txt").read_text() == "previous"
+    assert (custom / "note.txt").read_text() == "custom"
+
+
+def test_failed_rollback_keeps_interrupted_skill_outside_temporary_snapshots(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    skill = tmp_path / "browser-skill"
+    skill.mkdir()
+    (skill / "note.txt").write_text("Previous installed version\n")
+    with tempfile.TemporaryDirectory(dir=tmp_path) as temporary:
+        snapshots = windows_install._snapshot_paths([skill], Path(temporary))
+        (skill / "note.txt").write_text("Customization made during installation\n")
+
+        def fail_copy(source: Path, target: Path) -> None:
+            raise OSError("Injected restore failure")
+
+        monkeypatch.setattr(windows_install, "_copy_path", fail_copy)
+        with pytest.raises(Exception, match="Injected restore failure") as caught:
+            windows_install._restore_paths(snapshots)
+
+    retained = [
+        path
+        for path in tmp_path.rglob("note.txt")
+        if path.read_text() == "Customization made during installation\n"
+    ]
+    assert len(retained) == 1, "A restore failure must retain the interrupted skill"
+    assert str(retained[0].parent) in str(caught.value)
+
+
+@pytest.mark.parametrize("previously_installed", [False, True])
+def test_declared_skill_path_keeps_regular_file_replacement(
+    tmp_path: Path,
+    previously_installed: bool,
+) -> None:
+    skills = tmp_path / "skills"
+    skills.mkdir()
+    skill = skills / PACKAGED_HOST_SKILL_IDS[0]
+    skill_paths = windows_install._skill_mutation_paths(skills)
+    assert skill in skill_paths
+    if previously_installed:
+        skill.mkdir()
+        (skill / "SKILL.md").write_text("Previous installed version\n")
+    with tempfile.TemporaryDirectory(dir=tmp_path) as temporary:
+        snapshots = windows_install._snapshot_paths([skill], Path(temporary))
+        if skill.exists():
+            shutil.rmtree(skill)
+        skill.write_text("Concurrent replacement file\n")
+        retained = windows_install._restore_paths(snapshots, skill_paths=skill_paths)
+
+    if previously_installed:
+        assert (skill / "SKILL.md").read_text() == "Previous installed version\n"
+    else:
+        assert not skill.exists()
+    assert len(retained) == 1
+    assert retained[0].read_text() == "Concurrent replacement file\n"
+
+
+@pytest.mark.parametrize(
+    ("previously_installed", "parent_exists"),
+    [(False, False), (False, True), (True, True)],
+)
+def test_missing_skill_restores_snapshot_and_remaining_paths(
+    tmp_path: Path, previously_installed: bool, parent_exists: bool
+) -> None:
+    skill = tmp_path / "skills" / "browser-skill"
+    pointer = tmp_path / "current-release.json"
+    pointer.write_bytes(b"previous pointer\n")
+    if parent_exists:
+        skill.parent.mkdir()
+    if previously_installed:
+        skill.mkdir()
+        (skill / "SKILL.md").write_bytes(b"previous skill\n")
+    with tempfile.TemporaryDirectory(dir=tmp_path) as temporary:
+        # Recovery visits the missing skill before the pointer.
+        snapshots = windows_install._snapshot_paths([pointer, skill], Path(temporary))
+        pointer.write_bytes(b"interrupted pointer\n")
+        if previously_installed:
+            shutil.rmtree(skill)
+        retained = windows_install._restore_paths(snapshots, skill_paths=[skill])
+
+    assert pointer.read_bytes() == b"previous pointer\n"
+    if previously_installed:
+        assert (skill / "SKILL.md").read_bytes() == b"previous skill\n"
+    else:
+        assert not skill.exists()
+    assert retained == [], "An absent target has no interrupted content to retain"
+    assert not list(tmp_path.rglob("*.install-recovery.*"))
+
+
+def test_skill_rollback_retains_dangling_replacement_link(tmp_path: Path) -> None:
+    skill = tmp_path / "browser-skill"
+    skill.mkdir()
+    (skill / "SKILL.md").write_bytes(b"previous skill\n")
+    with tempfile.TemporaryDirectory(dir=tmp_path) as temporary:
+        snapshots = windows_install._snapshot_paths([skill], Path(temporary))
+        shutil.rmtree(skill)
+        try:
+            skill.symlink_to("missing-customization", target_is_directory=True)
+        except OSError:
+            pytest.skip("The host does not permit directory symlinks")
+        retained = windows_install._restore_paths(snapshots, skill_paths=[skill])
+
+    assert (skill / "SKILL.md").read_bytes() == b"previous skill\n"
+    assert len(retained) == 1 and retained[0].is_symlink()
+    assert os.readlink(retained[0]) == "missing-customization"
+
+
+@pytest.mark.parametrize("previously_installed", [False, True])
+def test_windows_installer_restores_after_skill_copy_fails_before_creation(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    previously_installed: bool,
+) -> None:
+    source = tmp_path / "source"
+    install_root = tmp_path / "install"
+    skills = tmp_path / "codex" / "skills"
+    skill = skills / PACKAGED_HOST_SKILL_IDS[0]
+    candidate = install_root / "releases" / "copy-failure"
+    source_skill = source / "skills" / skill.name / "SKILL.md"
+    source_skill.parent.mkdir(parents=True)
+    source_skill.write_bytes(b"candidate skill\n")
+    launcher_source = source / "scripts" / "loopx.ps1"
+    launcher_source.parent.mkdir()
+    launcher_source.write_bytes(b"candidate launcher\n")
+    previous = {
+        install_root / "current-release.json": b"previous pointer\n",
+        tmp_path / "bin" / "loopx.ps1": b"previous launcher\n",
+        tmp_path / "bin" / windows_install.LAUNCHER_POINTER_FILENAME: b"previous sidecar\n",
+    }
+    if previously_installed:
+        previous[skill / "SKILL.md"] = b"previous skill\n"
+    for path, content in previous.items():
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_bytes(content)
+
+    # Exercise the installer filesystem transaction on every host. Native
+    # Windows/PowerShell admission and candidate qualification remain separate.
+    if os.name != "nt":
+        monkeypatch.setattr(
+            windows_install, "os",
+            SimpleNamespace(name="nt", environ=os.environ, replace=os.replace, getpid=os.getpid),
+        )
+    monkeypatch.setattr(windows_install, "_validate_candidate", lambda *args, **kwargs: None)
+    monkeypatch.setattr(windows_install, "_upgrade_authority_archive", lambda *args, **kwargs: None)
+    copytree = shutil.copytree
+    attempted: list[Path] = []
+
+    def fail_skill_copy(source_path: Any, target: Any, *args: Any, **kwargs: Any) -> Any:
+        if Path(target) == skill and Path(source_path) == candidate / "skills" / skill.name:
+            assert not skill.exists(), "The real installer already removed the old skill"
+            attempted.append(Path(target))
+            raise OSError("Injected skill copy failure before target creation")
+        return copytree(source_path, target, *args, **kwargs)
+
+    monkeypatch.setattr(shutil, "copytree", fail_skill_copy)
+    with pytest.raises(Exception) as caught:
+        windows_install.install_windows(
+            source_root=source,
+            install_root=install_root,
+            bin_dir=tmp_path / "bin",
+            skills_dir=skills,
+            python_requested=sys.executable,
+            install_skills=True,
+            requested_release_id="copy-failure",
+        )
+
+    assert attempted == [skill]
+    assert not list(install_root.glob(".install-rollback.*"))
+    for path, content in previous.items():
+        assert path.read_bytes() == content, "Rollback must survive temporary snapshot cleanup"
+    if not previously_installed:
+        assert not skill.exists()
+    assert (candidate / "skills" / skill.name / "SKILL.md").read_bytes() == b"candidate skill\n"
+    assert isinstance(caught.value, OSError), "Successful rollback preserves the original error"
+    assert "Injected skill copy failure" in str(caught.value)
+
+
+def test_regular_file_rollback_retains_existing_launcher_behavior(
+    tmp_path: Path,
+) -> None:
+    launcher = tmp_path / "loopx.ps1"
+    launcher.write_text("Previous launcher\n")
+    with tempfile.TemporaryDirectory(dir=tmp_path) as temporary:
+        snapshots = windows_install._snapshot_paths([launcher], Path(temporary))
+        launcher.write_text("Interrupted candidate launcher\n")
+        windows_install._restore_paths(snapshots)
+    assert launcher.read_text() == "Previous launcher\n"
+    assert sorted(tmp_path.iterdir()) == [launcher]
+
+
 @pytest.mark.skipif(os.name != "nt", reason="native Windows installer regression")
 def test_windows_installer_promotes_release_and_runs_doctor(tmp_path: Path) -> None:
     pwsh = shutil.which("pwsh")