Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .github/workflows/python-tests.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
19 changes: 19 additions & 0 deletions docs/guides/installing-loopx.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
10 changes: 7 additions & 3 deletions loopx/python_install_owner.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,19 +30,23 @@ 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
path as a string literal, so the ``&`` call operator with a single-quoted
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(
Expand Down
56 changes: 55 additions & 1 deletion loopx/self_update.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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(
Expand All @@ -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)
Expand Down
14 changes: 14 additions & 0 deletions tests/test_python_install_owner_platform.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""

Expand Down
111 changes: 111 additions & 0 deletions tests/test_self_update_runtime_activation.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,8 @@
import sys
import threading
import time
import zipfile
import zipimport
from unittest import mock

from loopx import __version__
Expand Down Expand Up @@ -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"}
Expand Down
Loading