diff --git a/ares_runtime/local_runtime.py b/ares_runtime/local_runtime.py index a6d5d6cec11fc..6aceee2bc679e 100644 --- a/ares_runtime/local_runtime.py +++ b/ares_runtime/local_runtime.py @@ -36,6 +36,47 @@ class AresLocalRuntimeError(RuntimeError): """Raised when the explicit local-runtime contract is not satisfied.""" +def _prepare_gateway_stop_marker() -> None: + """Publish stop intent through the gateway owner in its process home.""" + from gateway import status + + pid_path = status._get_pid_path() + try: + identity = status.get_running_pid_identity_strict(pid_path) + except (OSError, RuntimeError) as exc: + raise AresLocalRuntimeError( + f"Ares gateway stop identity is ambiguous: {exc}" + ) from exc + if identity is None: + return + home = pid_path.parent + records = ( + status._read_pid_record(pid_path), + status._read_gateway_lock_record(status._get_gateway_lock_path(pid_path)), + ) + for record in records: + recorded_home = record.get("hermes_home") + if ( + not isinstance(recorded_home, str) + or not recorded_home.strip() + or not status._same_hermes_home(recorded_home, home) + or not status._record_matches_live_gateway_pid( + record, identity[0], expected_home=home + ) + ): + raise AresLocalRuntimeError("Ares gateway stop identity has an ambiguous home") + try: + current = status.get_running_pid_identity_strict(pid_path) + except (OSError, RuntimeError) as exc: + raise AresLocalRuntimeError( + f"Ares gateway stop identity changed: {exc}" + ) from exc + if current != identity: + raise AresLocalRuntimeError("Ares gateway stop identity changed before marker publication") + if not status.write_planned_stop_marker(identity[0]): + raise AresLocalRuntimeError("Ares gateway planned-stop marker could not be written") + + def _desktop_launch_arguments( executable: Path, *, @@ -1455,12 +1496,49 @@ def tui(self, arguments: Sequence[str]) -> None: def chat(self, arguments: Sequence[str]) -> None: self._exec_hermes(arguments) + def _prepare_gateway_stop(self) -> None: + # Gateway identity/marker paths deliberately ignore task-local home + # overrides. Scope their canonical owner in a child without changing + # the caller's process environment or active profile context. + _, source = self.active_release() + python = self._python_for(source) + environment = self._agent_environment() + environment["HERMES_HOME"] = str(self.paths.agent_home.expanduser().resolve()) + environment["PYTHONDONTWRITEBYTECODE"] = "1" + try: + completed = subprocess.run( + [ + str(python), + "-c", + "from ares_runtime.local_runtime import _prepare_gateway_stop_marker; " + "_prepare_gateway_stop_marker()", + ], + cwd=source, + env=environment, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=False, + timeout=10, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise AresLocalRuntimeError( + f"Ares gateway stop preparation failed: {exc}" + ) from exc + if completed.returncode: + detail = (completed.stderr or completed.stdout).strip() + raise AresLocalRuntimeError( + "Ares gateway stop preparation failed" + + (f": {detail}" if detail else "") + ) + def gateway(self, action: str) -> None: if action == "foreground": self._exec_hermes(["gateway"]) if action == "start": self._systemctl("enable", "--now", "ares-gateway.service") elif action == "stop": + self._prepare_gateway_stop() self._systemctl("disable", "--now", "ares-gateway.service") elif action == "restart": self._systemctl("restart", "ares-gateway.service") diff --git a/tests/ares_runtime/test_local_runtime.py b/tests/ares_runtime/test_local_runtime.py index 2f44a94605ee8..4f0c0592a65ec 100644 --- a/tests/ares_runtime/test_local_runtime.py +++ b/tests/ares_runtime/test_local_runtime.py @@ -61,6 +61,225 @@ def _repository(path: Path) -> Path: return path +@pytest.fixture +def gateway_stop_case(tmp_path: Path, monkeypatch): + from gateway import status + + runtime = _runtime(tmp_path) + source = _release(runtime, "a" * 40) + python = source / ".venv" / "bin" / "python" + python.parent.mkdir(parents=True) + python.touch() + runtime._activate("a" * 40) + home = runtime.paths.agent_home + home.mkdir() + pid, start = 987654, 1200 + record = { + "pid": pid, + "start_time": start, + "kind": "hermes-gateway", + "argv": ["python", "-m", "hermes_cli.main", "gateway", "run"], + "hermes_home": str(home), + } + for name in ("gateway.pid", "gateway.lock"): + (home / name).write_text(json.dumps(record)) + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setattr(status, "_is_gateway_runtime_lock_active_strict", lambda _path: True) + monkeypatch.setattr(status, "_pid_exists", lambda candidate: candidate == pid) + monkeypatch.setattr(status, "_get_process_start_time", lambda _pid: start) + monkeypatch.setattr(status, "_read_process_cmdline", lambda _pid: " ".join(record["argv"])) + return SimpleNamespace(runtime=runtime, source=source, home=home, + status=status, pid=pid, start=start, record=record) + + +def _gateway_stop_child(case, monkeypatch, events): + import ares_runtime.local_runtime as local_runtime + + def run(command, **kwargs): + assert command[:2] == [str(case.source / ".venv/bin/python"), "-c"] + assert kwargs["cwd"] == case.source + assert kwargs["env"]["HERMES_HOME"] == str(case.home.resolve()) + assert kwargs["timeout"] == 10 + events.append("prepare") + # Emulate the child's process environment without launching a process. + with monkeypatch.context() as child: + child.setenv("HERMES_HOME", kwargs["env"]["HERMES_HOME"]) + try: + exec(command[-1], {}) + except AresLocalRuntimeError as exc: + return SimpleNamespace(returncode=1, stdout="", stderr=str(exc)) + return SimpleNamespace(returncode=0, stdout="", stderr="") + + monkeypatch.setattr(local_runtime.subprocess, "run", run) + monkeypatch.setattr(case.runtime, "_systemctl", + lambda *args, **kwargs: events.append(args) or True) + + +def test_gateway_stop_publishes_verified_marker_before_service_stop(gateway_stop_case, monkeypatch): + case = gateway_stop_case + events = [] + _gateway_stop_child(case, monkeypatch, events) + case.runtime.gateway("stop") + + assert events == ["prepare", ("disable", "--now", "ares-gateway.service")] + marker = json.loads((case.home / ".gateway-planned-stop.json").read_text()) + assert (marker["target_pid"], marker["target_start_time"]) == (case.pid, case.start) + with monkeypatch.context() as consumer: + consumer.setattr(case.status.os, "getpid", lambda: case.pid) + assert case.status.consume_planned_stop_marker_for_self() is True + + +def test_gateway_stop_keeps_isolated_unit_routing(gateway_stop_case, monkeypatch): + from dataclasses import replace + import ares_runtime.local_runtime as local_runtime + + case = gateway_stop_case + case.runtime = AresLocalRuntime(replace(case.runtime.paths, unit_path=case.home / "offline-stop.service")) + events = [] + _gateway_stop_child(case, monkeypatch, events) + child_run = local_runtime.subprocess.run + monkeypatch.setattr(case.runtime, "_systemctl", AresLocalRuntime._systemctl.__get__(case.runtime)) + monkeypatch.setattr(local_runtime.shutil, "which", lambda _name: "/usr/bin/systemctl") + + def run(command, **kwargs): + if command[0] != "systemctl": + return child_run(command, **kwargs) + assert (case.home / ".gateway-planned-stop.json").is_file() + events.append(command) + return SimpleNamespace(returncode=0, stdout="", stderr="") + + monkeypatch.setattr(local_runtime.subprocess, "run", run) + case.runtime.gateway("stop") + assert events == ["prepare", ["systemctl", "--user", "disable", "--now", "offline-stop.service"]] + + +@pytest.mark.parametrize("boundary", ["stale_start", "dead_pid", "wrong_pid", "wrong_role"]) +def test_gateway_stop_rejects_stale_or_disagreeing_identity(gateway_stop_case, monkeypatch, boundary): + case = gateway_stop_case + if boundary == "stale_start": + monkeypatch.setattr(case.status, "_get_process_start_time", lambda _pid: case.start + 1) + elif boundary == "dead_pid": + monkeypatch.setattr(case.status, "_pid_exists", lambda _pid: False) + elif boundary == "wrong_pid": + record = dict(case.record, pid=case.pid + 1) + (case.home / "gateway.lock").write_text(json.dumps(record)) + else: + monkeypatch.setattr(case.status, "_read_process_cmdline", lambda _pid: "python -m hermes_cli.main serve") + events = [] + _gateway_stop_child(case, monkeypatch, events) + with pytest.raises(AresLocalRuntimeError, match="identity"): + case.runtime.gateway("stop") + assert events == ["prepare"] + assert not (case.home / ".gateway-planned-stop.json").exists() + + +@pytest.mark.parametrize("record_name", ["gateway.pid", "gateway.lock", "live_profile"]) +def test_gateway_stop_rejects_foreign_home(gateway_stop_case, monkeypatch, tmp_path, record_name): + case = gateway_stop_case + if record_name == "live_profile": + monkeypatch.setattr(case.status, "_read_process_cmdline", lambda _pid: "python -m hermes_cli.main --profile other gateway run") + else: + record = dict(case.record, hermes_home=str(tmp_path / "foreign-home")) + (case.home / record_name).write_text(json.dumps(record)) + events = [] + _gateway_stop_child(case, monkeypatch, events) + with pytest.raises(AresLocalRuntimeError, match="home"): + case.runtime.gateway("stop") + assert events == ["prepare"] + assert not (case.home / ".gateway-planned-stop.json").exists() + + +def test_gateway_stop_rejects_identity_change_before_publication(gateway_stop_case, monkeypatch): + case = gateway_stop_case + identities = iter(((case.pid, case.start), (case.pid, case.start + 1))) + monkeypatch.setattr(case.status, "get_running_pid_identity_strict", lambda _path: next(identities)) + events = [] + _gateway_stop_child(case, monkeypatch, events) + with pytest.raises(AresLocalRuntimeError, match="identity"): + case.runtime.gateway("stop") + assert events == ["prepare"] + assert not (case.home / ".gateway-planned-stop.json").exists() + + +def test_gateway_stop_aborts_when_marker_publication_fails(gateway_stop_case, monkeypatch): + case = gateway_stop_case + monkeypatch.setattr(case.status, "write_planned_stop_marker", lambda _pid: False) + events = [] + _gateway_stop_child(case, monkeypatch, events) + with pytest.raises(AresLocalRuntimeError, match="marker"): + case.runtime.gateway("stop") + assert events == ["prepare"] + + +def test_gateway_stop_without_a_runtime_owner_does_not_create_marker(gateway_stop_case, monkeypatch): + case = gateway_stop_case + (case.home / "gateway.pid").unlink() + (case.home / "gateway.lock").unlink() + events = [] + _gateway_stop_child(case, monkeypatch, events) + case.runtime.gateway("stop") + assert events == ["prepare", ("disable", "--now", "ares-gateway.service")] + assert not (case.home / ".gateway-planned-stop.json").exists() + + +def test_gateway_stop_preserves_caller_environment_and_profile_context(gateway_stop_case, monkeypatch, tmp_path): + from hermes_constants import get_hermes_home, reset_hermes_home_override, set_hermes_home_override + + case = gateway_stop_case + ambient_home = tmp_path / "ambient-home" + monkeypatch.setenv("HERMES_HOME", str(ambient_home)) + monkeypatch.setenv("PYTHONPATH", "ambient-python-path") + profile = tmp_path / "context-profile" + token = set_hermes_home_override(profile) + before = dict(os.environ) + try: + events = [] + _gateway_stop_child(case, monkeypatch, events) + case.runtime.gateway("stop") + assert dict(os.environ) == before + assert get_hermes_home() == profile + assert (case.home / ".gateway-planned-stop.json").is_file() + assert not (ambient_home / ".gateway-planned-stop.json").exists() + assert not (profile / ".gateway-planned-stop.json").exists() + finally: + reset_hermes_home_override(token) + + +@pytest.mark.parametrize("failure", [OSError("synthetic launch failure"), subprocess.TimeoutExpired("synthetic", 10)]) +def test_gateway_stop_preparation_failure_never_calls_service(gateway_stop_case, monkeypatch, failure): + case = gateway_stop_case + before = dict(os.environ) + monkeypatch.setattr("ares_runtime.local_runtime.subprocess.run", + lambda *args, **kwargs: (_ for _ in ()).throw(failure)) + calls = [] + monkeypatch.setattr(case.runtime, "_systemctl", lambda *args, **kwargs: calls.append(args)) + with pytest.raises(AresLocalRuntimeError, match="stop"): + case.runtime.gateway("stop") + assert calls == [] + assert dict(os.environ) == before + + +@pytest.mark.parametrize("mismatch", ["stale", "pid", "start", "home"]) +def test_gateway_stop_marker_owner_rejects_stale_or_wrong_consumer(gateway_stop_case, monkeypatch, tmp_path, mismatch): + import ares_runtime.local_runtime as local_runtime + + case = gateway_stop_case + local_runtime._prepare_gateway_stop_marker() + path = case.home / ".gateway-planned-stop.json" + marker = json.loads(path.read_text()) + if mismatch == "stale": + marker["written_at"] = "2000-01-01T00:00:00+00:00" + elif mismatch == "pid": + marker["target_pid"] += 1 + elif mismatch == "start": + marker["target_start_time"] += 1 + else: + marker["target_hermes_home"] = str(tmp_path / "foreign-consumer") + path.write_text(json.dumps(marker)) + monkeypatch.setattr(case.status.os, "getpid", lambda: case.pid) + assert case.status.consume_planned_stop_marker_for_self() is False + + def test_current_link_is_the_only_active_runtime_pointer(tmp_path: Path) -> None: runtime = _runtime(tmp_path) first = "a" * 40