fix(update): rename exe on Windows instead of detached process

src/conductor/__init__.py

@@ -30,6 +30,8 @@
     exceptions: Custom exception hierarchy.
 """
 
-__version__ = "0.1.0"
+from importlib.metadata import version as _pkg_version
+
+__version__ = _pkg_version("conductor-cli")
 
 __all__ = ["__version__"]

src/conductor/cli/update.py

@@ -13,8 +13,10 @@
 
 from __future__ import annotations
 
+import contextlib
 import json
 import logging
+import shutil
 import subprocess
 import sys
 import urllib.error
@@ -273,6 +275,18 @@ def _print_hint(console: Console, remote_version: str) -> None:
 # ---------------------------------------------------------------------------
 
 
+def _get_conductor_exe() -> Path | None:
+    """Return the path to the ``conductor`` executable, or ``None`` if not found.
+
+    Uses :func:`shutil.which` to locate the executable on ``$PATH``.
+
+    Returns:
+        A :class:`Path` to the executable, or ``None``.
+    """
+    which = shutil.which("conductor")
+    return Path(which) if which else None
+
+
 def run_update(console: Console) -> None:
     """Fetch the latest version and self-upgrade via ``uv tool install``.
 
@@ -301,31 +315,35 @@ def run_update(console: Console) -> None:
     install_url = f"git+{_REPO_GIT_URL}@{tag_name}"
     cmd = ["uv", "tool", "install", "--force", install_url]
 
+    # On Windows, rename our exe out of the way so uv can write the new one.
+    # Windows locks running executables but allows renaming them.
+    old_exe: Path | None = None
     if sys.platform == "win32":
-        # Windows locks running executables — spawn uv detached and exit
-        # so conductor.exe releases its file lock before uv overwrites it.
-        DETACHED_PROCESS = 0x00000008
-        CREATE_NEW_PROCESS_GROUP = 0x00000200
-        subprocess.Popen(  # noqa: S603
-            cmd,
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
-            creationflags=DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP,
-        )
-        console.print(f"Upgrade to v{version} started in background. It will complete momentarily.")
-        # Clear cache so next run re-checks
-        cache_path = get_cache_path()
-        cache_path.unlink(missing_ok=True)
-        return
+        exe_path = _get_conductor_exe()
+        if exe_path and exe_path.exists():
+            old_exe = exe_path.with_suffix(".exe.old")
+            # Clean up leftover .old from a previous successful update
+            if old_exe.exists():
+                with contextlib.suppress(OSError):
+                    old_exe.unlink()
+            try:
+                exe_path.rename(old_exe)
+            except OSError:
+                old_exe = None  # rename failed; proceed anyway, uv will report the error
 
     proc = subprocess.run(cmd, capture_output=True, text=True)  # noqa: S603
 
     if proc.returncode == 0:
         console.print(f"[green]Successfully upgraded to v{version}[/green]")
-        # Clear cache so next run re-checks
         cache_path = get_cache_path()
         cache_path.unlink(missing_ok=True)
     else:
         console.print(f"[bold red]Upgrade failed[/bold red] (exit code {proc.returncode})")
         if proc.stderr:
             console.print(f"[dim]{proc.stderr.strip()}[/dim]")
+        # On Windows, restore the original exe if uv failed to write a new one
+        if old_exe and old_exe.exists():
+            exe_path = old_exe.with_suffix("")  # .exe.old → .exe
+            if not exe_path.exists():
+                with contextlib.suppress(OSError):
+                    old_exe.rename(exe_path)

tests/test_cli/test_update.py

@@ -505,48 +505,108 @@ def test_command_includes_tag_name(self, cache_dir: Path) -> None:
         assert len(install_arg) == 1
         assert install_arg[0].endswith("@v2.0.0")
 
-    def test_windows_uses_detached_process(self, cache_dir: Path) -> None:
-        """On Windows, ``run_update`` spawns a detached ``Popen`` and exits early."""
-        import subprocess as sp
+    def test_windows_renames_exe_before_install(self, cache_dir: Path, tmp_path: Path) -> None:
+        """On Windows, ``run_update`` renames the exe to ``.exe.old`` before calling ``uv``."""
+        # Create a fake conductor.exe
+        fake_exe = tmp_path / "conductor.exe"
+        fake_exe.write_text("fake")
 
         cache_file = cache_dir / "update-check.json"
         cache_file.write_text("{}")
 
         c, buf = _make_console(is_terminal=True)
+        mock_proc = MagicMock()
+        mock_proc.returncode = 0
+        mock_proc.stderr = ""
 
         with (
             patch(
                 "conductor.cli.update.fetch_latest_version",
                 return_value=("99.0.0", "v99.0.0", "https://example.com"),
             ),
             patch("conductor.cli.update.sys.platform", "win32"),
-            patch("conductor.cli.update.subprocess.Popen") as mock_popen,
-            patch("conductor.cli.update.subprocess.run") as mock_run,
+            patch("conductor.cli.update._get_conductor_exe", return_value=fake_exe),
+            patch("conductor.cli.update.subprocess.run", return_value=mock_proc) as mock_run,
         ):
             run_update(c)
 
-        # Popen must be called with DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP
-        mock_popen.assert_called_once()
-        call_kwargs = mock_popen.call_args[1]
-        DETACHED_PROCESS = 0x00000008
-        CREATE_NEW_PROCESS_GROUP = 0x00000200
-        expected_flags = DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP
-        assert call_kwargs["creationflags"] == expected_flags
-        assert call_kwargs["stdout"] == sp.DEVNULL
-        assert call_kwargs["stderr"] == sp.DEVNULL
+        # The exe should have been renamed to .exe.old
+        old_exe = tmp_path / "conductor.exe.old"
+        assert old_exe.exists()
 
-        # subprocess.run must NOT be called on Windows
-        mock_run.assert_not_called()
+        # subprocess.run must be called (not Popen)
+        mock_run.assert_called_once()
 
-        # Output should mention background
+        # Output should say successful (synchronous path)
         output = buf.getvalue()
-        assert "started in background" in output
+        assert "Successfully upgraded" in output
 
         # Cache should be cleared
         assert not cache_file.exists()
 
-    def test_unix_uses_synchronous_subprocess(self, cache_dir: Path) -> None:
-        """On non-Windows platforms, ``run_update`` uses synchronous ``subprocess.run``."""
+    def test_windows_restores_exe_on_failure(self, cache_dir: Path, tmp_path: Path) -> None:
+        """On Windows, if ``uv`` fails and doesn't write a new exe, the old one is restored."""
+        # Create a fake conductor.exe
+        fake_exe = tmp_path / "conductor.exe"
+        fake_exe.write_text("fake")
+
+        c, buf = _make_console(is_terminal=True)
+        mock_proc = MagicMock()
+        mock_proc.returncode = 1
+        mock_proc.stderr = "install failed"
+
+        with (
+            patch(
+                "conductor.cli.update.fetch_latest_version",
+                return_value=("99.0.0", "v99.0.0", "https://example.com"),
+            ),
+            patch("conductor.cli.update.sys.platform", "win32"),
+            patch("conductor.cli.update._get_conductor_exe", return_value=fake_exe),
+            patch("conductor.cli.update.subprocess.run", return_value=mock_proc),
+        ):
+            run_update(c)
+
+        # The .old file should have been renamed back to .exe since uv didn't write a new one
+        assert fake_exe.exists()
+        old_exe = tmp_path / "conductor.exe.old"
+        assert not old_exe.exists()
+
+        # Output should report failure
+        output = buf.getvalue()
+        assert "Upgrade failed" in output
+
+    def test_windows_cleans_up_previous_old_exe(self, cache_dir: Path, tmp_path: Path) -> None:
+        """On Windows, a leftover ``.exe.old`` from a previous update is deleted first."""
+        # Create a fake conductor.exe and a pre-existing .old file
+        fake_exe = tmp_path / "conductor.exe"
+        fake_exe.write_text("new-fake")
+        old_leftover = tmp_path / "conductor.exe.old"
+        old_leftover.write_text("stale-old")
+
+        c, buf = _make_console(is_terminal=True)
+        mock_proc = MagicMock()
+        mock_proc.returncode = 0
+        mock_proc.stderr = ""
+
+        with (
+            patch(
+                "conductor.cli.update.fetch_latest_version",
+                return_value=("99.0.0", "v99.0.0", "https://example.com"),
+            ),
+            patch("conductor.cli.update.sys.platform", "win32"),
+            patch("conductor.cli.update._get_conductor_exe", return_value=fake_exe),
+            patch("conductor.cli.update.subprocess.run", return_value=mock_proc),
+        ):
+            run_update(c)
+
+        # The old leftover should be gone, replaced by the newly renamed exe
+        old_exe = tmp_path / "conductor.exe.old"
+        assert old_exe.exists()
+        # The content should be "new-fake" (current exe), not "stale-old"
+        assert old_exe.read_text() == "new-fake"
+
+    def test_unix_skips_rename(self, cache_dir: Path) -> None:
+        """On non-Windows platforms, ``_get_conductor_exe`` is not called."""
         c, buf = _make_console(is_terminal=True)
         mock_proc = MagicMock()
         mock_proc.returncode = 0
@@ -559,17 +619,17 @@ def test_unix_uses_synchronous_subprocess(self, cache_dir: Path) -> None:
             ),
             patch("conductor.cli.update.sys.platform", "linux"),
             patch("conductor.cli.update.subprocess.run", return_value=mock_proc) as mock_run,
-            patch("conductor.cli.update.subprocess.Popen") as mock_popen,
+            patch("conductor.cli.update._get_conductor_exe") as mock_get_exe,
         ):
             run_update(c)
 
-        # subprocess.run must be called on Linux
-        mock_run.assert_called_once()
+        # _get_conductor_exe must NOT be called on Linux
+        mock_get_exe.assert_not_called()
 
-        # Popen must NOT be called on Linux
-        mock_popen.assert_not_called()
+        # subprocess.run must be called
+        mock_run.assert_called_once()
 
-        # Output should mention successful upgrade (synchronous path)
+        # Output should mention successful upgrade
         output = buf.getvalue()
         assert "Successfully upgraded" in output