diff --git a/.github/workflows/python-tests.yml b/.github/workflows/python-tests.yml index 81e84efdc3..abcdd12001 100644 --- a/.github/workflows/python-tests.yml +++ b/.github/workflows/python-tests.yml @@ -863,6 +863,7 @@ jobs: tests/test_file_lock_cross_process.py tests/control_plane/test_effect_runtime_integration.py tests/control_plane/test_shadow_observation_retirement.py + tests/test_python_install_owner_platform.py tests/test_self_update_runtime_activation.py tests/test_self_update_download.py tests/test_windows_atomic_writes.py diff --git a/docs/guides/installing-loopx.md b/docs/guides/installing-loopx.md index d82581dcba..9b75df2390 100644 --- a/docs/guides/installing-loopx.md +++ b/docs/guides/installing-loopx.md @@ -395,6 +395,25 @@ processes to install workflow skills and slash commands, runs doctor, revalidate enabled extensions, and restarts managed local LoopX services. It does not switch the installation to an archive snapshot. +On Windows, an update requested through the pip/pipx `loopx.exe` launcher is +refused before installation starts: the running launcher would prevent pip +from replacing it and can leave a partially uninstalled package. Follow the +returned command, which uses the owning Python interpreter's module entry and +preserves the original arguments, including registry and runtime routes: + +```powershell +python -m loopx.entrypoint --format json update apply +``` + +Linked pipx application aliases, such as `loopx-preview.exe` installed with +`--suffix=-preview`, receive the same protection when they refer to the owning +environment's `loopx.exe` file. + +Use the interpreter printed in the refusal rather than an unrelated `python` +from PATH. Read-only `update check` and `update plan` remain available. If a +previous update already damaged the installation, repair it with that owning +interpreter's pip first; a module entry cannot import a missing package. + The equivalent manual sequence is: ```bash diff --git a/loopx/python_install_owner.py b/loopx/python_install_owner.py index bb6090996a..41e388867e 100644 --- a/loopx/python_install_owner.py +++ b/loopx/python_install_owner.py @@ -30,7 +30,7 @@ def _is_windows_host() -> bool: return os.name == "nt" -def interpreter_invocation(python_executable: str) -> str: +def interpreter_invocation(python_executable: str, arguments: list[str] | None = None) -> str: """The text that runs an interpreter path on the host that prints it. POSIX shells run a quoted path directly. PowerShell reads a bare quoted @@ -38,11 +38,15 @@ def interpreter_invocation(python_executable: str) -> str: literal is required; that form also carries spaces and apostrophes in the interpreter path. This module owns the renderer because the pip-owned upgrade command is the first caller; doctor and update reuse it. + Arguments, when supplied, use the same host quoting and retain empty values. """ if _is_windows_host(): - return f"& {_powershell_literal(python_executable)}" - return shlex.quote(python_executable) + return " ".join( + [f"& {_powershell_literal(python_executable)}", + *(_powershell_literal(value) for value in arguments or [])] + ) + return shlex.join([python_executable, *(arguments or [])]) def resolve_python_install_owner( diff --git a/loopx/self_update.py b/loopx/self_update.py index 74e291a7b7..5522a54b11 100644 --- a/loopx/self_update.py +++ b/loopx/self_update.py @@ -9,6 +9,7 @@ import shlex import subprocess import sys +import zipimport from typing import Any from urllib.parse import quote from urllib.request import Request, urlopen @@ -508,6 +509,55 @@ def _python_distribution_rollback_plan( } +def _windows_console_update_refusal(payload: dict[str, Any]) -> dict[str, Any] | None: + """A distlib launcher holds its own executable open until Python exits. + + Use its actual zip loader, not argv[0]: distlib removes the .exe suffix. + The module entry has a source loader and can safely replace the launcher. + """ + loader = getattr(sys.modules.get("__main__"), "__loader__", None) + lifecycle = payload.get("install_lifecycle") or {} + if ( + sys.platform != "win32" + or not isinstance(loader, zipimport.zipimporter) + or lifecycle.get("execution_driver") not in {"python_pip", "python_pipx"} + ): + return None + launcher = Path(loader.archive) + if launcher.name.casefold() != "loopx.exe": + # pipx suffixes the exposed app name, but its symlink/hardlink can still + # hold this installation's actual console executable open. + owned_launcher = Path(sys.executable).with_name("loopx.exe") + try: + if not launcher.samefile(owned_launcher): + return None + except FileNotFoundError: + return None + retry = interpreter_invocation(sys.executable, ["-m", "loopx.entrypoint", *sys.argv[1:]]) + reason = ( + "The running Windows console launcher cannot be replaced by pip. " + "No update was started; retry through the owning Python interpreter." + ) + return { + **payload, + "ok": False, + "changes_applied": False, + "execution": None, + "error": reason, + "recommended_action": f"{reason}\n{retry}", + "install_lifecycle": {**lifecycle, "loopx_apply_supported": False, "reason": reason}, + "plan": { + **payload.get("plan", {}), "apply_supported": False, + "blocked_reason": reason, "mutates_release_install": False, + }, + "commands": {**payload.get("commands", {}), "apply": retry}, + "next_action": { + "kind": "use_installation_owner", "command": retry, + "mutating": True, "requires_explicit_approval": True, "reason": reason, + }, + } + + def build_rollback_plan( *, release_id: str, @@ -831,7 +881,7 @@ def build_update_plan( "requires_explicit_approval": bool(owner_upgrade_command), "reason": recommended_action, } - return { + payload = { "ok": not execute or apply_supported, "schema_version": UPDATE_PLAN_SCHEMA_VERSION, "mode": "update", @@ -893,6 +943,7 @@ def build_update_plan( ), "recommended_action": recommended_action, } + return (_windows_console_update_refusal(payload) if execute else None) or payload def _execute_python_distribution_update( @@ -902,6 +953,9 @@ def _execute_python_distribution_update( route_arguments: list[str], progress: Callable[[str], None] = lambda _stage: None, ) -> dict[str, Any]: + refused = _windows_console_update_refusal(payload) + if refused is not None: + return refused lifecycle = ( payload.get("install_lifecycle") if isinstance(payload.get("install_lifecycle"), dict) diff --git a/tests/test_python_install_owner_platform.py b/tests/test_python_install_owner_platform.py index d5834af2bb..b4e2af2055 100644 --- a/tests/test_python_install_owner_platform.py +++ b/tests/test_python_install_owner_platform.py @@ -60,6 +60,20 @@ def pip_owner_upgrade_command() -> str: return command +def test_windows_interpreter_arguments_round_trip_through_powershell( + monkeypatch: pytest.MonkeyPatch, +) -> None: + if PWSH is None: + pytest.skip("PowerShell is unavailable") + pretend_windows_host(monkeypatch) + arguments = ["-c", "import json,sys; print(json.dumps(sys.argv[1:]))", "O'Brien $name", "δΈ­ζ–‡ πŸš€", "", 'a"b'] + command = python_install_owner.interpreter_invocation(sys.executable, arguments) + completed = subprocess.run([PWSH, "-NoLogo", "-NoProfile", "-Command", command], + capture_output=True, encoding="utf-8", check=True) + import json + assert json.loads(completed.stdout) == arguments[2:] + + def pip_doctor_payload() -> dict[str, object]: """Just enough collected-doctor shape for a pip-owned update plan.""" diff --git a/tests/test_self_update_runtime_activation.py b/tests/test_self_update_runtime_activation.py index 020b3f8207..0e018c8154 100644 --- a/tests/test_self_update_runtime_activation.py +++ b/tests/test_self_update_runtime_activation.py @@ -7,6 +7,8 @@ import sys import threading import time +import zipfile +import zipimport from unittest import mock from loopx import __version__ @@ -486,6 +488,115 @@ def test_python_distribution_apply_uses_the_owning_interpreter_pip() -> None: assert run.call_args_list[3].args[0][3:5] == ["slash-commands", "--install"] +@pytest.mark.parametrize("installer", ["pip", "pipx"]) +@pytest.mark.parametrize("linked_alias", [False, True]) +def test_windows_console_apply_refuses_before_installation( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, installer: str, linked_alias: bool, +) -> None: + doctor = doctor_payload() + doctor["package"] = {"install_kind": "python_distribution"} + doctor["install_freshness"]["python_distribution_installer"] = installer + # A saved plan must not bypass the check at execution time. + saved_plan = build_update_plan(action="apply", doctor_payload=doctor) + launcher = tmp_path / "loopx.exe" + with zipfile.ZipFile(launcher, "w") as archive: + archive.writestr("__main__.py", "") + if linked_alias: + alias = tmp_path / "loopx-preview.exe" + alias.hardlink_to(launcher) + launcher = alias + monkeypatch.setattr(sys, "executable", str(tmp_path / "python.exe")) + monkeypatch.setattr(sys, "platform", "win32") + monkeypatch.setattr(sys.modules["__main__"], "__loader__", zipimport.zipimporter(str(launcher))) + monkeypatch.setattr(sys, "argv", [str(launcher.with_suffix("")), "--registry", "project O'Brien $name/registry.json", + "--format", "json", "update", "--execute", "--timeout-seconds", "60"]) + with mock.patch("loopx.self_update.subprocess.run") as run: + planned = build_update_plan(action="apply", doctor_payload=doctor) + executed = execute_update_plan(saved_plan) + run.assert_not_called() + for result in (planned, executed): + assert result["ok"] is False + assert result["changes_applied"] is False + assert result["execution"] is None + assert result["plan"]["apply_supported"] is False + assert result["plan"]["mutates_release_install"] is False + assert "launcher" in result["error"].lower() + retry = result["next_action"]["command"] + assert "loopx.entrypoint" in retry + assert "--registry" in retry and "--timeout-seconds" in retry + assert result["next_action"]["mutating"] is True + + +@pytest.mark.parametrize("owned_exists", [False, True]) +def test_windows_unrelated_zip_console_keeps_update_plan( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, owned_exists: bool, +) -> None: + doctor = doctor_payload() + doctor["package"] = {"install_kind": "python_distribution"} + doctor["install_freshness"]["python_distribution_installer"] = "pipx" + owned = tmp_path / "loopx.exe" + with zipfile.ZipFile(owned, "w") as archive: + archive.writestr("__main__.py", "") + unrelated = tmp_path / "unrelated.exe" + unrelated.write_bytes(owned.read_bytes()) + if not owned_exists: + owned.unlink() + monkeypatch.setattr(sys, "platform", "win32") + monkeypatch.setattr(sys, "executable", str(tmp_path / "python.exe")) + baseline = build_update_plan(action="apply", doctor_payload=doctor) + monkeypatch.setattr(sys.modules["__main__"], "__loader__", zipimport.zipimporter(str(unrelated))) + assert build_update_plan(action="apply", doctor_payload=doctor) == baseline + + +@pytest.mark.parametrize("action", ["check", "plan"]) +def test_windows_console_read_only_update_keeps_its_plan( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, action: str, +) -> None: + doctor = doctor_payload() + doctor["package"] = {"install_kind": "python_distribution"} + doctor["install_freshness"]["python_distribution_installer"] = "pip" + baseline = build_update_plan(action=action, doctor_payload=doctor) + launcher = tmp_path / "loopx.exe" + with zipfile.ZipFile(launcher, "w") as archive: + archive.writestr("__main__.py", "") + monkeypatch.setattr(sys, "platform", "win32") + monkeypatch.setattr(sys.modules["__main__"], "__loader__", zipimport.zipimporter(str(launcher))) + assert build_update_plan(action=action, doctor_payload=doctor) == baseline + + +@pytest.mark.skipif(os.name != "nt", reason="native Windows console launcher") +def test_real_windows_console_refuses_before_starting_installer(tmp_path: Path) -> None: + from distlib.scripts import ScriptMaker + + marker = tmp_path / "installer-started" + (tmp_path / "update_probe.py").write_text( + "import json,subprocess\n" + "from pathlib import Path\n" + "from loopx import self_update\n" + "def main():\n" + " def installer(*args, **kwargs):\n" + f" Path({str(marker)!r}).write_text('unsafe installation', encoding='utf-8')\n" + " return subprocess.CompletedProcess([], 1, '', 'installation refused by fixture')\n" + " self_update.subprocess.run = installer\n" + " payload = {'install_lifecycle': {'execution_driver': 'python_pip'}, 'plan': {'apply_supported': True}}\n" + " result = self_update.execute_update_plan(payload)\n" + " print(json.dumps(result))\n" + " return 0 if result['ok'] else 1\n", + encoding="utf-8", + ) + maker = ScriptMaker(None, str(tmp_path)) + maker.variants = {""} + maker.make("loopx = update_probe:main") + env = {**os.environ, "PYTHONPATH": os.pathsep.join([str(tmp_path), str(REPO_ROOT)])} + result = subprocess.run([str(tmp_path / "loopx.exe"), "update", "apply"], + env=env, capture_output=True, encoding="utf-8", timeout=30) + assert result.returncode == 1, result.stderr + assert not marker.exists(), "the real launcher reached the installer" + refusal = json.loads(result.stdout) + assert refusal["execution"] is None + assert "loopx.entrypoint" in refusal["next_action"]["command"] + + def test_pipx_distribution_apply_preserves_the_pipx_environment() -> None: doctor = doctor_payload() doctor["package"] = {"install_kind": "python_distribution"}