Skip to content
Merged
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
50 changes: 49 additions & 1 deletion catalog/desktop/probe.py
Original file line number Diff line number Diff line change
Expand Up @@ -3581,14 +3581,61 @@ def _resolve_profile_home(name: str, here: Optional[Path] = None) -> Optional[Pa
return candidate.resolve() if candidate.is_dir() else None


def _hermes_install_pythons() -> list[str]:
"""Interpreters of installer-managed Hermes dependency environments, active first.

The default installer starts the gateway on a bare bundled interpreter
(``~/.hermes/tools/python-*/bin/python3 -I``), so neither its argv[0] nor a
``python3`` on the gateway's PATH imports httpx. Its packages live in a
generated venv, ``<HERMES_HOME>/installs/<id>/environments/<hash>/venv``,
whose hash changes on every update. ``installs/<id>/facts.json`` names the
active one (``packages.venv.environment``). Other environments follow,
newest first, for an install whose facts are missing or unreadable. Named
profiles live under ``<base>/profiles/<name>``; installs stay in the base.
"""
interpreter = Path("Scripts/python.exe") if os.name == "nt" else Path("bin/python")
roots: list[Path] = []
for home in _hermes_homes():
roots.append(home)
if home.parent.name == "profiles":
roots.append(home.parent.parent)
active: list[str] = []
others: list[tuple[float, str]] = []
for root in dict.fromkeys(roots):
installs = root / "installs"
try:
if not installs.is_dir():
continue
facts_files = sorted(installs.glob("*/facts.json"))
venvs = list(installs.glob("*/environments/*/venv"))
except OSError:
continue
for facts in facts_files:
try:
environment = json.loads(facts.read_text(encoding="utf-8"))["packages"]["venv"]["environment"]
except (OSError, ValueError, KeyError, TypeError):
continue
if isinstance(environment, str) and Path(environment).is_absolute():
active.append(str(Path(environment) / interpreter))
for venv in venvs:
try:
others.append(((venv / "pyvenv.cfg").stat().st_mtime, str(venv / interpreter)))
except OSError:
continue
ordered = active + [path for _, path in sorted(others, reverse=True)]
return [path for path in dict.fromkeys(ordered) if Path(path).is_file()]


def _gateway_python_candidates(proc_root: Path = Path("/proc")) -> list[str]:
"""Discover on this backend, without resolving venv interpreter symlinks.

On Linux shell.exec is a child of the Python gateway. Walk only our own
ancestors, reading argv[0] (never returning or logging command arguments).
/proc/<pid>/exe would lose the venv by resolving to the system binary.
Other platforms and restricted proc mounts retain the environment and
Desktop's existing home/PATH fallbacks.
Desktop's existing home/PATH fallbacks. A gateway from the default
installer runs a bare bundled interpreter; its dependency venv comes last,
from the installer's own records (``_hermes_install_pythons``).
"""
candidates: list[str] = []
explicit = os.environ.get("HERMES_PYTHON", "").strip()
Expand All @@ -3615,6 +3662,7 @@ def _gateway_python_candidates(proc_root: Path = Path("/proc")) -> list[str]:
pid = int(parent.group(1))
except (OSError, ValueError):
break
candidates.extend(_hermes_install_pythons())
return list(dict.fromkeys(candidates))


Expand Down
50 changes: 49 additions & 1 deletion probe.py
Original file line number Diff line number Diff line change
Expand Up @@ -3913,14 +3913,61 @@ def _resolve_profile_home(name: str, here: Optional[Path] = None) -> Optional[Pa
return candidate.resolve() if candidate.is_dir() else None


def _hermes_install_pythons() -> list[str]:
"""Interpreters of installer-managed Hermes dependency environments, active first.

The default installer starts the gateway on a bare bundled interpreter
(``~/.hermes/tools/python-*/bin/python3 -I``), so neither its argv[0] nor a
``python3`` on the gateway's PATH imports httpx. Its packages live in a
generated venv, ``<HERMES_HOME>/installs/<id>/environments/<hash>/venv``,
whose hash changes on every update. ``installs/<id>/facts.json`` names the
active one (``packages.venv.environment``). Other environments follow,
newest first, for an install whose facts are missing or unreadable. Named
profiles live under ``<base>/profiles/<name>``; installs stay in the base.
"""
interpreter = Path("Scripts/python.exe") if os.name == "nt" else Path("bin/python")
roots: list[Path] = []
for home in _hermes_homes():
roots.append(home)
if home.parent.name == "profiles":
roots.append(home.parent.parent)
active: list[str] = []
others: list[tuple[float, str]] = []
for root in dict.fromkeys(roots):
installs = root / "installs"
try:
if not installs.is_dir():
continue
facts_files = sorted(installs.glob("*/facts.json"))
venvs = list(installs.glob("*/environments/*/venv"))
except OSError:
continue
for facts in facts_files:
try:
environment = json.loads(facts.read_text(encoding="utf-8"))["packages"]["venv"]["environment"]
except (OSError, ValueError, KeyError, TypeError):
continue
if isinstance(environment, str) and Path(environment).is_absolute():
active.append(str(Path(environment) / interpreter))
for venv in venvs:
try:
others.append(((venv / "pyvenv.cfg").stat().st_mtime, str(venv / interpreter)))
except OSError:
continue
ordered = active + [path for _, path in sorted(others, reverse=True)]
return [path for path in dict.fromkeys(ordered) if Path(path).is_file()]


def _gateway_python_candidates(proc_root: Path = Path("/proc")) -> list[str]:
"""Discover on this backend, without resolving venv interpreter symlinks.

On Linux shell.exec is a child of the Python gateway. Walk only our own
ancestors, reading argv[0] (never returning or logging command arguments).
/proc/<pid>/exe would lose the venv by resolving to the system binary.
Other platforms and restricted proc mounts retain the environment and
Desktop's existing home/PATH fallbacks.
Desktop's existing home/PATH fallbacks. A gateway from the default
installer runs a bare bundled interpreter; its dependency venv comes last,
from the installer's own records (``_hermes_install_pythons``).
"""
candidates: list[str] = []
explicit = os.environ.get("HERMES_PYTHON", "").strip()
Expand All @@ -3947,6 +3994,7 @@ def _gateway_python_candidates(proc_root: Path = Path("/proc")) -> list[str]:
pid = int(parent.group(1))
except (OSError, ValueError):
break
candidates.extend(_hermes_install_pythons())
return list(dict.fromkeys(candidates))


Expand Down
97 changes: 94 additions & 3 deletions test_probe_runtime.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,26 @@ def probes():
MISSING = {"provider": "resetwatch", "details": ["probe cannot import httpx: No module named 'httpx'"]}


def interpreter(probe, venv):
return str(venv / ("Scripts/python.exe" if probe.os.name == "nt" else "bin/python"))


def make_install_venv(probe, home, install_id, environment, *, mtime):
"""A dependency venv laid out like the default Hermes installer's."""
venv = home / "installs" / install_id / "environments" / environment / "venv"
python = Path(interpreter(probe, venv))
python.parent.mkdir(parents=True)
python.write_text("")
(venv / "pyvenv.cfg").write_text("home = /bundled/python/bin\n")
os.utime(venv / "pyvenv.cfg", (mtime, mtime))
return venv


def write_facts(home, install_id, venv):
facts = {"schema": 1, "packages": {"venv": {"environment": str(venv), "extras": ["all"]}}}
(home / "installs" / install_id / "facts.json").write_text(json.dumps(facts))


class ProbeRuntimeTests(unittest.TestCase):
def test_dependency_failure_is_not_a_complete_collection(self):
for source, probe in probes():
Expand Down Expand Up @@ -62,14 +82,16 @@ def test_parent_runtime_keeps_venv_path_and_ignores_process_arguments(self):
folder.mkdir()
(folder / "cmdline").write_bytes(b"\0".join(arg.encode() for arg in argv))
(folder / "status").write_text(f"Name:\tprocess\nPPid:\t{parent}\n")
with patch.dict(os.environ, {}, clear=True), patch.object(probe.os, "getppid", return_value=30):
with patch.dict(os.environ, {}, clear=True), patch.object(probe.os, "getppid", return_value=30), \
patch.object(probe, "_hermes_homes", return_value=[]):
self.assertEqual(probe._gateway_python_candidates(proc), ["/opt/custom install/venv/bin/python3"])

def test_backend_overrides_and_missing_proc(self):
for source, probe in probes():
with self.subTest(source=source), tempfile.TemporaryDirectory() as tmp:
runtime = str(Path(tmp) / "custom python")
with patch.dict(os.environ, {"HERMES_PYTHON": runtime}, clear=True):
with patch.dict(os.environ, {"HERMES_PYTHON": runtime}, clear=True), \
patch.object(probe, "_hermes_homes", return_value=[]):
self.assertEqual(probe._gateway_python_candidates(Path(tmp)), [runtime])

def test_container_gateway_at_pid_one_is_discovered(self):
Expand All @@ -79,9 +101,78 @@ def test_container_gateway_at_pid_one_is_discovered(self):
folder.mkdir()
(folder / "cmdline").write_bytes(b"/usr/local/lib/hermes-agent/venv/bin/python3\0-m\0tui_gateway")
(folder / "status").write_text("PPid:\t0\n")
with patch.dict(os.environ, {}, clear=True), patch.object(probe.os, "getppid", return_value=1):
with patch.dict(os.environ, {}, clear=True), patch.object(probe.os, "getppid", return_value=1), \
patch.object(probe, "_hermes_homes", return_value=[]):
self.assertEqual(probe._gateway_python_candidates(Path(tmp)), ["/usr/local/lib/hermes-agent/venv/bin/python3"])

def test_default_installer_venv_comes_from_install_facts(self):
# The default installer runs the gateway on a bare bundled interpreter
# and keeps httpx in a generated venv that facts.json names (#35).
for source, probe in probes():
with self.subTest(source=source), tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
home = root / ".hermes"
# /proc argv[0] is a POSIX path; a temp path would be C:\... on Windows.
bare = "/home/user/.hermes/tools/python-3.14.7/bin/python3"
proc = root / "proc"
folder = proc / "20"
folder.mkdir(parents=True)
(folder / "cmdline").write_bytes(bare.encode() + b"\0-I\0-c\0launcher")
(folder / "status").write_text("PPid:\t0\n")
active = make_install_venv(probe, home, "a1b2", "f2cc", mtime=1_000)
stale = make_install_venv(probe, home, "a1b2", "0ld0", mtime=2_000)
write_facts(home, "a1b2", active)
with patch.dict(os.environ, {}, clear=True), patch.object(probe.os, "getppid", return_value=20), \
patch.object(probe, "_hermes_homes", return_value=[home]):
self.assertEqual(probe._gateway_python_candidates(proc),
[bare, interpreter(probe, active), interpreter(probe, stale)])

def test_dependency_venvs_without_facts_are_tried_newest_first(self):
for source, probe in probes():
with self.subTest(source=source), tempfile.TemporaryDirectory() as tmp:
home = Path(tmp) / ".hermes"
older = make_install_venv(probe, home, "a1b2", "1111", mtime=1_000)
newer = make_install_venv(probe, home, "a1b2", "2222", mtime=2_000)
broken = make_install_venv(probe, home, "a1b2", "3333", mtime=3_000)
Path(interpreter(probe, broken)).unlink()
with patch.dict(os.environ, {}, clear=True), patch.object(probe.os, "getppid", return_value=0), \
patch.object(probe, "_hermes_homes", return_value=[home]):
self.assertEqual(probe._gateway_python_candidates(Path(tmp) / "proc"),
[interpreter(probe, newer), interpreter(probe, older)])

def test_named_profile_home_uses_the_base_installs(self):
for source, probe in probes():
with self.subTest(source=source), tempfile.TemporaryDirectory() as tmp:
base = Path(tmp) / ".hermes"
active = make_install_venv(probe, base, "a1b2", "f2cc", mtime=1_000)
write_facts(base, "a1b2", active)
with patch.dict(os.environ, {}, clear=True), patch.object(probe.os, "getppid", return_value=0), \
patch.object(probe, "_hermes_homes", return_value=[base / "profiles" / "work"]):
self.assertEqual(probe._gateway_python_candidates(Path(tmp) / "proc"), [interpreter(probe, active)])

def test_gateway_runtime_reexecs_from_bare_interpreter_into_installer_venv(self):
class ReplacedProcess(BaseException):
pass

for source, probe in probes():
with self.subTest(source=source), tempfile.TemporaryDirectory() as tmp:
home = Path(tmp) / ".hermes"
active = make_install_venv(probe, home, "a1b2", "f2cc", mtime=1_000)
write_facts(home, "a1b2", active)
flags = [source, "--profile", "default", "--pin-snapshot", "--slice=0:8"]
with patch.dict(os.environ, {}, clear=True), patch.object(probe.os, "getppid", return_value=0), \
patch.object(probe, "_hermes_homes", return_value=[home]), \
patch.object(probe.sys, "argv", flags), \
patch.dict("sys.modules", {"httpx": None}), \
patch.object(probe.subprocess, "run", return_value=SimpleNamespace(returncode=0)) as check, \
patch.object(probe.os, "execv", side_effect=ReplacedProcess) as replace:
with self.assertRaises(ReplacedProcess):
probe._use_gateway_python()
self.assertEqual(check.call_args.args[0], [interpreter(probe, active), "-c", "import httpx"])
replace.assert_called_once_with(
interpreter(probe, active),
[interpreter(probe, active), str(Path(probe.__file__).absolute()), *flags[1:]])

def test_failed_candidates_fall_back_and_reexec_preserves_profile_flags(self):
class ReplacedProcess(BaseException):
pass
Expand Down
Loading