diff --git a/loopx/command_invocation.py b/loopx/command_invocation.py index 6375cf8a43..7e202a9a76 100644 --- a/loopx/command_invocation.py +++ b/loopx/command_invocation.py @@ -43,13 +43,38 @@ def cli_option_args(flag: str, value: object) -> list[str]: def current_loopx_cli_argv() -> list[str]: - """Pin captured readers to this installed interpreter or explicit CLI. + """Pin captured readers to this installed release or explicit CLI. A native Host can run from a project outside the source checkout. An ambient - ``loopx`` may be another installation and lack this capture's reader. + ``loopx`` may be another installation and lack this capture's reader. The + installed launcher also preserves the release root that made this module + importable; invoking its Python without that launcher can load a stale module. """ route = loopx_cli_argv() - return [sys.executable, "-m", "loopx.entrypoint"] if route == ["loopx"] else route + if route != ["loopx"]: + return route + + release_root_text = os.environ.get("LOOPX_RELEASE_ROOT", "").strip() + if release_root_text: + release_root: Path | None = None + module_root: Path | None = None + launcher: Path | None = None + launcher_ready = False + try: + release_root = Path(release_root_text).expanduser().resolve() + module_root = Path(__file__).resolve().parents[1] + launcher = release_root / "scripts" / ( + "loopx.ps1" if os.name == "nt" else "loopx" + ) + launcher_ready = launcher.is_file() and ( + os.name == "nt" or os.access(launcher, os.X_OK) + ) + except (OSError, RuntimeError): + pass + if release_root == module_root and launcher is not None and launcher_ready: + return loopx_cli_argv(str(launcher)) + + return [sys.executable, "-m", "loopx.entrypoint"] def resolve_command_path( diff --git a/tests/control_plane/test_replan_staged_read.py b/tests/control_plane/test_replan_staged_read.py index f8cee18e4d..c5da60a849 100644 --- a/tests/control_plane/test_replan_staged_read.py +++ b/tests/control_plane/test_replan_staged_read.py @@ -611,6 +611,18 @@ def test_native_reader_keeps_its_installation_across_acceptance_and_writeback(tm from loopx.control_plane.turn_driver.decision_capture import write_decision_capture from loopx.control_plane.turn_driver.host_candidate import extract_turn_authority, render_prompt + release_root = Path(__file__).resolve().parents[2] + release_cli = release_root / "scripts" / "loopx" + stale_root = tmp_path / "stale-system-module" + stale_package = stale_root / "loopx" + stale_package.mkdir(parents=True) + (stale_package / "__init__.py").write_text("") + (stale_package / "entrypoint.py").write_text( + "raise SystemExit('stale system LoopX must not handle captured readers')\n" + ) + monkeypatch.setenv("LOOPX_RELEASE_ROOT", str(release_root)) + monkeypatch.setenv("PYTHONPATH", str(stale_root)) + source = _quota_payload() outcome = {"id": "original-outcome", "objective": "Deliver the original accepted artifact.", "requirements": [{"id": "original-stop", "description": "Stop before publication."}]} @@ -623,7 +635,9 @@ def test_native_reader_keeps_its_installation_across_acceptance_and_writeback(tm args = {"captured_decision_path": str(capture), "turn_instance_id": "original-native-turn", "turn_key": "sha256:" + "0" * 64} legacy = bind_turn_decision_capture(source, envelope, **args) assert "read autonomous_replan_obligation.replan_context" in render_prompt(extract_turn_authority({"turn_envelope": legacy})) - bound = bind_turn_decision_capture(source, envelope, **args, reader_argv=current_loopx_cli_argv()) + reader_argv = current_loopx_cli_argv() + assert reader_argv == [str(release_cli), "--cli-bin", str(release_cli)] + bound = bind_turn_decision_capture(source, envelope, **args, reader_argv=reader_argv) authority = extract_turn_authority({"turn_envelope": bound}) trap = tmp_path / "bin" trap.mkdir() diff --git a/tests/test_command_invocation.py b/tests/test_command_invocation.py index 9af06f91df..4f2542e90e 100644 --- a/tests/test_command_invocation.py +++ b/tests/test_command_invocation.py @@ -7,7 +7,11 @@ import pytest -from loopx.command_invocation import command_argv, resolve_command_path +from loopx.command_invocation import ( + command_argv, + current_loopx_cli_argv, + resolve_command_path, +) def test_loopx_route_is_one_argument_and_does_not_leak_on_failure(): @@ -23,6 +27,29 @@ def test_loopx_route_is_one_argument_and_does_not_leak_on_failure(): assert loopx_cli_argv() == ["loopx"] +def test_current_cli_uses_exact_active_release_launcher(monkeypatch): + import loopx.command_invocation as invocation + + release_root = Path(invocation.__file__).resolve().parents[1] + launcher = release_root / "scripts" / ("loopx.ps1" if os.name == "nt" else "loopx") + monkeypatch.setenv("LOOPX_RELEASE_ROOT", str(release_root)) + + assert current_loopx_cli_argv() == [ + str(launcher), "--cli-bin", str(launcher), + ] + + +def test_current_cli_ignores_foreign_release_root(tmp_path, monkeypatch): + launcher = tmp_path / "scripts" / ("loopx.ps1" if os.name == "nt" else "loopx") + launcher.parent.mkdir() + launcher.write_text("foreign release") + monkeypatch.setenv("LOOPX_RELEASE_ROOT", str(tmp_path)) + + assert current_loopx_cli_argv() == [ + __import__("sys").executable, "-m", "loopx.entrypoint", + ] + + @pytest.mark.parametrize("value", ["", " ", "-invalid", "name\nnext", "name\x00next"]) def test_invalid_cli_route_is_rejected_before_dispatch(value): from loopx.cli_runtime import build_cli_parser