diff --git a/loopx/windows_install.py b/loopx/windows_install.py index 5a69957af2..f753955309 100644 --- a/loopx/windows_install.py +++ b/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 { diff --git a/tests/test_windows_install.py b/tests/test_windows_install.py index 4a9e5b30ec..d90690c1b2 100644 --- a/tests/test_windows_install.py +++ b/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")