Prevent self-upgrades from nesting the managed payload

docs/operations/install-binaries.md

@@ -147,10 +147,12 @@ The canonical install is the one this document describes: a launcher at
 `~/.local/bin/kickstart` (symlink or `#!/bin/sh` wrapper), owned by
 `kickstart install`, which leads to the active managed payload under
 `~/.local/share/kickstart`. `kickstart upgrade` refreshes that managed
-installation. Treat the payload's internal path as an implementation detail
-and use `kickstart install --check` to inspect it. Nothing else should shadow
-the launcher. In particular, ad-hoc shims in personal `bin` directories (for
-example
+installation while reusing the same launcher and managed app root on every
+run. A successful upgrade also collapses an accidentally nested payload left
+by older upgrade logic back into that stable root. Treat the payload's
+internal path as an implementation detail and use `kickstart install --check`
+to inspect it. Nothing else should shadow the launcher. In particular, ad-hoc
+shims in personal `bin` directories (for example
 `~/workspace/bin/kickstart`) are not supported: they bypass `kickstart
 upgrade`, go stale silently, and hide which binary owns the command. Delete
 them, or reduce them to a one-line `exec "$HOME/.local/bin/kickstart" "$@"`

src/utils/installer.py

@@ -64,17 +64,36 @@ class PathUpdateResult:
     on_path_now: bool
 
 
+def current_entrypoint_path() -> Path:
+    """Return the absolute invocation path without resolving symlinks.
+
+    Preserving the logical path matters for managed installs: the launcher
+    symlink identifies both its public directory and the managed app root. A
+    bare command name is resolved through ``PATH`` so callers still receive an
+    absolute path.
+    """
+    raw_argv0 = sys.argv[0] if sys.argv and sys.argv[0] else None
+    if raw_argv0 is not None:
+        argv0 = Path(raw_argv0).expanduser()
+        candidate = argv0
+        is_bare_command = os.sep not in raw_argv0 and (os.altsep is None or os.altsep not in raw_argv0)
+        if is_bare_command:
+            located = shutil.which(str(argv0))
+            if located is not None:
+                candidate = Path(located)
+        if candidate.exists() or candidate.is_symlink():
+            return Path(os.path.abspath(candidate))
+    return Path(os.path.abspath(sys.executable))
+
+
 def current_binary_path() -> Path:
-    """Return the absolute path of the running kickstart entry point.
+    """Return the resolved absolute path of the running kickstart executable.
 
     Tries `sys.argv[0]` first (correct for PyInstaller `--onefile` and `--onedir`
     builds and for console-script shims) and falls back to `sys.executable` when
     argv[0] does not point to a real file (this is what happens with `python -m`).
     """
-    argv0 = Path(sys.argv[0]) if sys.argv and sys.argv[0] else None
-    if argv0 is not None and argv0.exists():
-        return _safe_resolve(argv0)
-    return Path(sys.executable).resolve()
+    return _safe_resolve(current_entrypoint_path())
 
 
 def detect_shell(env: Optional[dict[str, str]] = None) -> Optional[str]:
@@ -212,21 +231,9 @@ def _install_onedir_bundle(
 
     copied = False
     if not _same_file(bundle_source, bundle_destination):
-        if bundle_destination.exists() or bundle_destination.is_symlink():
-            if not overwrite:
-                raise FileExistsError(f"{bundle_destination} already exists; pass --force to overwrite")
-            _remove_path(bundle_destination)
-
-        temp_destination = resolved_app_root / f".{APP_DIR_NAME}.tmp-{os.getpid()}"
-        if temp_destination.exists() or temp_destination.is_symlink():
-            _remove_path(temp_destination)
-        try:
-            shutil.copytree(bundle_source, temp_destination, symlinks=True)
-            temp_destination.rename(bundle_destination)
-        except Exception:
-            if temp_destination.exists() or temp_destination.is_symlink():
-                _remove_path(temp_destination)
-            raise
+        if (bundle_destination.exists() or bundle_destination.is_symlink()) and not overwrite:
+            raise FileExistsError(f"{bundle_destination} already exists; pass --force to overwrite")
+        _replace_bundle(bundle_source, bundle_destination)
         copied = True
 
     launcher_changed = _install_launcher(executable_destination, destination, overwrite=overwrite)
@@ -333,6 +340,43 @@ def _onedir_bundle_root(source: Path, *, name: str = BINARY_NAME) -> Optional[Pa
     return None
 
 
+def _replace_bundle(bundle_source: Path, bundle_destination: Path) -> None:
+    """Stage and activate a bundle with rollback for staging or activation errors."""
+    app_root = bundle_destination.parent
+    suffix = os.getpid()
+    temp_destination = app_root / f".{APP_DIR_NAME}.tmp-{suffix}"
+    backup_destination = app_root / f".{APP_DIR_NAME}.backup-{suffix}"
+
+    for path in (temp_destination, backup_destination):
+        if path.exists() or path.is_symlink():
+            _remove_path(path)
+
+    try:
+        shutil.copytree(bundle_source, temp_destination, symlinks=True)
+    except Exception:
+        if temp_destination.exists() or temp_destination.is_symlink():
+            _remove_path(temp_destination)
+        raise
+
+    previous_moved = False
+    try:
+        if bundle_destination.exists() or bundle_destination.is_symlink():
+            bundle_destination.rename(backup_destination)
+            previous_moved = True
+        try:
+            temp_destination.rename(bundle_destination)
+        except Exception:
+            if previous_moved:
+                backup_destination.rename(bundle_destination)
+            raise
+    finally:
+        if temp_destination.exists() or temp_destination.is_symlink():
+            _remove_path(temp_destination)
+
+    if previous_moved:
+        _remove_path(backup_destination)
+
+
 def _install_launcher(executable: Path, destination: Path, *, overwrite: bool) -> bool:
     """Expose `executable` at `destination`, preferring a symlink."""
     if destination.exists() or destination.is_symlink():

src/utils/updater.py

@@ -31,7 +31,7 @@
     APP_DIR_NAME,
     BINARY_NAME,
     InstallResult,
-    current_binary_path,
+    current_entrypoint_path,
     default_app_root,
     install_binary,
 )
@@ -96,21 +96,34 @@ def find_asset(assets: list[ReleaseAsset], name: str) -> Optional[ReleaseAsset]:
 def resolve_current_install_layout() -> tuple[Path, Optional[Path]]:
     """Return ``(launcher_dir, app_root_or_None)`` for the currently running install.
 
-    When the launcher is a symlink (the onedir install shape), we can recover
-    the app root from the symlink target. For legacy single-file installs the
-    launcher is a regular file and we return ``None`` so callers fall back to
-    `default_app_root` (or skip app-root handling entirely).
+    A managed launcher symlink reveals the public launcher directory and app
+    root through its first target. A wrapper-launched managed payload reveals
+    the stable app root directly. For legacy single-file installs we return
+    ``None`` so callers fall back to `default_app_root` (or skip app-root
+    handling entirely).
     """
-    launcher = current_binary_path()
+    launcher = current_entrypoint_path()
     launcher_dir = launcher.parent
 
+    # A generated wrapper executes the stable managed payload directly. In
+    # that case argv[0] is already <app_root>/current/kickstart, and refreshing
+    # that directory in place keeps the external wrapper valid.
+    if (
+        launcher.name == BINARY_NAME
+        and launcher.parent.name == APP_DIR_NAME
+        and (launcher.parent / "_internal").is_dir()
+    ):
+        return launcher_dir, launcher.parent.parent
+
     if launcher.is_symlink():
         try:
-            executable = launcher.resolve()
+            target = launcher.readlink()
         except OSError:
             return launcher_dir, None
+        executable = target if target.is_absolute() else launcher.parent / target
+        executable = Path(os.path.abspath(executable))
         # Expected layout: <app_root>/<APP_DIR_NAME>/<BINARY_NAME>
-        if executable.parent.name == APP_DIR_NAME:
+        if executable.name == BINARY_NAME and executable.parent.name == APP_DIR_NAME:
             return launcher_dir, executable.parent.parent
 
     return launcher_dir, None

tests/unit/utils/test_installer.py

@@ -70,6 +70,44 @@ def test_install_binary_missing_source_raises(tmp_path: Path) -> None:
         installer.install_binary(missing, tmp_path / "bin")
 
 
+def test_current_entrypoint_path_preserves_symlink_for_layout_discovery(tmp_path: Path, monkeypatch) -> None:
+    target = tmp_path / "app" / installer.BINARY_NAME
+    target.parent.mkdir()
+    target.write_text("#!/bin/sh\n")
+    launcher = tmp_path / "bin" / installer.BINARY_NAME
+    launcher.parent.mkdir()
+    launcher.symlink_to(target)
+    monkeypatch.setattr(installer.sys, "argv", [str(launcher)])
+
+    assert installer.current_entrypoint_path() == launcher
+    assert installer.current_binary_path() == target
+
+
+def test_current_entrypoint_path_finds_bare_command_on_path(tmp_path: Path, monkeypatch) -> None:
+    launcher = tmp_path / installer.BINARY_NAME
+    launcher.write_text("#!/bin/sh\n")
+    monkeypatch.setattr(installer.sys, "argv", [installer.BINARY_NAME])
+    monkeypatch.setattr(installer.shutil, "which", lambda command: str(launcher))
+
+    assert installer.current_entrypoint_path() == launcher
+
+
+def test_current_entrypoint_path_preserves_explicit_relative_path(tmp_path: Path, monkeypatch) -> None:
+    working_dir = tmp_path / "working"
+    working_dir.mkdir()
+    explicit_launcher = working_dir / installer.BINARY_NAME
+    explicit_launcher.write_text("#!/bin/sh\n")
+    path_launcher = tmp_path / "path-bin" / installer.BINARY_NAME
+    path_launcher.parent.mkdir()
+    path_launcher.write_text("#!/bin/sh\n")
+
+    monkeypatch.chdir(working_dir)
+    monkeypatch.setattr(installer.sys, "argv", [f".{os.sep}{installer.BINARY_NAME}"])
+    monkeypatch.setattr(installer.shutil, "which", lambda command: str(path_launcher))
+
+    assert installer.current_entrypoint_path() == explicit_launcher
+
+
 def test_uninstall_binary_removes_file(tmp_path: Path, single_file_binary: Path) -> None:
     target_dir = tmp_path / "user-bin"
     installer.install_binary(single_file_binary, target_dir)
@@ -469,6 +507,63 @@ def test_install_binary_onedir_overwrite_replaces_bundle(tmp_path: Path) -> None
     assert result.copied is True
 
 
+def test_install_binary_onedir_staging_failure_preserves_active_nested_bundle(tmp_path: Path, monkeypatch) -> None:
+    """A failed repair must leave the existing nested payload executable."""
+    target_dir = tmp_path / "bin"
+    target_dir.mkdir()
+    app_root = tmp_path / "share" / "kickstart"
+    bundle_dest = app_root / installer.APP_DIR_NAME
+    nested_bundle = bundle_dest / ".kickstart" / installer.APP_DIR_NAME
+    (nested_bundle / "_internal").mkdir(parents=True)
+    nested_executable = nested_bundle / installer.BINARY_NAME
+    nested_executable.write_text("#!/bin/sh\necho old\n")
+    nested_executable.chmod(0o755)
+
+    managed_executable = bundle_dest / installer.BINARY_NAME
+    managed_executable.symlink_to(nested_executable)
+    launcher = target_dir / installer.BINARY_NAME
+    launcher.symlink_to(managed_executable)
+    fresh_launcher = _make_fake_onedir_bundle(tmp_path / "fresh")
+
+    def fail_copytree(*args, **kwargs):
+        raise OSError("staging failed")
+
+    monkeypatch.setattr(installer.shutil, "copytree", fail_copytree)
+
+    with pytest.raises(OSError, match="staging failed"):
+        installer.install_binary(fresh_launcher, target_dir, app_root=app_root, overwrite=True)
+
+    assert launcher.resolve() == nested_executable
+    assert nested_executable.read_text() == "#!/bin/sh\necho old\n"
+
+
+def test_install_binary_onedir_activation_failure_restores_active_bundle(tmp_path: Path, monkeypatch) -> None:
+    """A failed activation rolls the old payload back into its stable path."""
+    old_launcher = _make_fake_onedir_bundle(tmp_path / "old")
+    target_dir = tmp_path / "bin"
+    app_root = tmp_path / "share" / "kickstart"
+    installer.install_binary(old_launcher, target_dir, app_root=app_root)
+    active_launcher = target_dir / installer.BINARY_NAME
+    old_target = active_launcher.resolve()
+    fresh_launcher = _make_fake_onedir_bundle(tmp_path / "fresh")
+    (fresh_launcher.parent / "_internal" / "marker").write_text("fresh\n")
+
+    original_rename = Path.rename
+
+    def fail_activation(self: Path, target: Path) -> Path:
+        if self.name.startswith(f".{installer.APP_DIR_NAME}.tmp-"):
+            raise OSError("activation failed")
+        return original_rename(self, target)
+
+    monkeypatch.setattr(Path, "rename", fail_activation)
+
+    with pytest.raises(OSError, match="activation failed"):
+        installer.install_binary(fresh_launcher, target_dir, app_root=app_root, overwrite=True)
+
+    assert active_launcher.resolve() == old_target
+    assert (app_root / installer.APP_DIR_NAME / "_internal" / "marker").read_text() == "present\n"
+
+
 def test_install_binary_onedir_idempotent_same_bundle(tmp_path: Path) -> None:
     """Re-installing the same bundle is a no-op for both the payload and launcher."""
     launcher = _make_fake_onedir_bundle(tmp_path / "src")
@@ -478,7 +573,7 @@ def test_install_binary_onedir_idempotent_same_bundle(tmp_path: Path) -> None:
 
     # Now re-install pointing at the freshly-installed bundle (the source IS the
     # destination after the first install).
-    new_source = (app_root / installer.APP_DIR_NAME / installer.BINARY_NAME)
+    new_source = app_root / installer.APP_DIR_NAME / installer.BINARY_NAME
     second_result = installer.install_binary(new_source, target_dir, app_root=app_root, overwrite=True)
     assert second_result.copied is False
     assert second_result.already_installed is True