Skip to content
Closed
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
44 changes: 40 additions & 4 deletions hermes_cli/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -7666,9 +7666,6 @@ def _detect_concurrent_hermes_instances(
except Exception:
return []

if exclude_pid is None:
exclude_pid = os.getpid()

# Resolve every shim path to its canonical form once for cheap comparison.
shim_paths: set[str] = set()
for shim in _hermes_exe_shims(scripts_dir):
Expand All @@ -7679,6 +7676,45 @@ def _detect_concurrent_hermes_instances(
if not shim_paths:
return []

# Build the exclusion set: our own PID plus every ancestor PID whose
# executable is one of our shims.
#
# The uv-generated ``hermes.exe`` is a launcher that spawns ``python.exe``
# as a child and waits for it. ``cmd_update`` runs inside the Python
# child, so ``os.getpid()`` is the child's PID — but the parent
# ``hermes.exe`` shim is alive the whole time and would otherwise be
# reported as "another running instance" with a fresh PID on every
# invocation (issue raised by Jeffrey: "hermes update fails saying it
# detects a running instance — gives me a new pid every time").
#
# Walking ancestors instead of just excluding the immediate parent is
# belt-and-suspenders for shells that may wrap hermes.exe through an
# extra layer (e.g. cmd /c hermes.exe inside a script).
if exclude_pid is None:
exclude_pid = os.getpid()
exclude_pids: set[int] = {int(exclude_pid)}
try:
proc = psutil.Process(int(exclude_pid))
ancestors = proc.parents()
except Exception:
ancestors = []
for ancestor in ancestors:
try:
anc_exe = ancestor.exe()
except Exception:
continue
if not anc_exe:
continue
try:
anc_norm = str(Path(anc_exe).resolve()).lower()
except (OSError, ValueError):
anc_norm = str(anc_exe).lower()
if anc_norm in shim_paths:
try:
exclude_pids.add(int(ancestor.pid))
except Exception:
continue

matches: list[tuple[int, str]] = []
try:
proc_iter = psutil.process_iter(["pid", "exe", "name"])
Expand All @@ -7692,7 +7728,7 @@ def _detect_concurrent_hermes_instances(
continue
pid = info.get("pid")
exe = info.get("exe")
if not exe or pid is None or pid == exclude_pid:
if not exe or pid is None or int(pid) in exclude_pids:
continue
try:
exe_norm = str(Path(exe).resolve()).lower()
Expand Down
86 changes: 86 additions & 0 deletions tests/hermes_cli/test_update_concurrent_quarantine.py
Original file line number Diff line number Diff line change
Expand Up @@ -118,6 +118,92 @@ def test_detect_concurrent_is_noop_off_windows(_winp, tmp_path):
assert cli_main._detect_concurrent_hermes_instances(tmp_path) == []


@patch.object(cli_main, "_is_windows", return_value=True)
def test_detect_concurrent_excludes_parent_shim(_winp, tmp_path):
"""Regression test: when ``cmd_update`` runs inside the Python child
spawned by the uv-generated ``hermes.exe`` shim, the parent shim must
NOT be reported as "another running instance" — that was the false
positive that caused ``hermes update`` to bail out with a fresh PID
on every invocation.

Setup:
- Our PID is C (the python child running cmd_update).
- C's parent is P (hermes.exe shim) — should be excluded.
- There's also an unrelated O (genuinely another hermes.exe) —
should still be reported.
"""
scripts_dir = tmp_path
shim = scripts_dir / "hermes.exe"
shim.write_bytes(b"")

my_pid = 27700
parent_pid = 4176
other_pid = 9999

# Fake psutil.Process(my_pid).parents() chain
parent_proc = MagicMock()
parent_proc.pid = parent_pid
parent_proc.exe.return_value = str(shim)

self_proc = MagicMock()
self_proc.parents.return_value = [parent_proc]

def fake_process_ctor(pid):
assert pid == my_pid
return self_proc

procs = [
_make_proc(parent_pid, str(shim), "hermes.exe"), # parent shim — exclude
_make_proc(my_pid, str(shim), "hermes.exe"), # self — exclude
_make_proc(other_pid, str(shim), "hermes.exe"), # genuine other instance
]
fake_psutil = types.SimpleNamespace(
process_iter=lambda attrs: iter(procs),
Process=fake_process_ctor,
)
with patch.dict(sys.modules, {"psutil": fake_psutil}):
result = cli_main._detect_concurrent_hermes_instances(
scripts_dir, exclude_pid=my_pid
)

assert result == [(other_pid, "hermes.exe")]


@patch.object(cli_main, "_is_windows", return_value=True)
def test_detect_concurrent_handles_ancestor_enum_failure(_winp, tmp_path):
"""If psutil can't enumerate ancestors (AccessDenied, NoSuchProcess,
or even a bare stub without ``Process``), we degrade gracefully and
still exclude at least the caller PID."""
scripts_dir = tmp_path
shim = scripts_dir / "hermes.exe"
shim.write_bytes(b"")

my_pid = 12345
other_pid = 67890
procs = [
_make_proc(my_pid, str(shim), "hermes.exe"), # self — exclude
_make_proc(other_pid, str(shim), "hermes.exe"), # genuine other
]

# Stub psutil that *raises* on Process() lookup — simulates the test
# environment in the rest of this file, where psutil is duck-typed
# to only provide process_iter.
def raising_process(_pid):
raise RuntimeError("simulated psutil failure")

fake_psutil = types.SimpleNamespace(
process_iter=lambda attrs: iter(procs),
Process=raising_process,
)
with patch.dict(sys.modules, {"psutil": fake_psutil}):
result = cli_main._detect_concurrent_hermes_instances(
scripts_dir, exclude_pid=my_pid
)

# Self is still excluded; the other process is still reported.
assert result == [(other_pid, "hermes.exe")]


# ---------------------------------------------------------------------------
# _format_concurrent_instances_message
# ---------------------------------------------------------------------------
Expand Down
Loading