hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 32 Packages Projects Releases Wiki Activity

refactor(cli-misc): loops.py _save helper; linux_desktop_entry.py shared head/shebang readers, probe via _run_quiet

Browse Source
This commit is contained in:
Teknium
2026-09-02 20:21:37 -07:00
parent fd55049657
commit cc369d088e
2 changed files with 54 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
hermes_cli/linux_desktop_entry.py
View File

@@ -75,18 +75,13 @@ def _can_import_hermes_cli(interpreter: Path) -> bool:
cached = _probe_cache.get(key)
if cached is not None:
return cached
try:
result = subprocess.run(
[key, "-I", "-c", "import hermes_cli.main"],
cwd=os.path.abspath(os.sep),
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
timeout=15,
)
except (OSError, subprocess.SubprocessError):
ok = _run_quiet(
[key, "-I", "-c", "import hermes_cli.main"],
cwd=os.path.abspath(os.sep), timeout=15, on_error=None,
)
if ok is None:
return True
_probe_cache[key] = ok = result.returncode == 0
_probe_cache[key] = ok
return ok
@@ -228,6 +223,18 @@ def _shebang_tokens(shebang: str) -> "list[str]":
return shebang[2:].strip().split()
def _is_native_binary(head: bytes) -> bool:
return head[:4] == b"\x7fELF" or head.startswith(b"MZ")
def _read_head(path: Path, size: int = 4096) -> Optional[bytes]:
try:
with open(path, "rb") as fh:
return fh.read(size)
except OSError:
return None
def _wrapper_shebang_safe(wrapper: Path) -> bool:
"""Whether an executable wrapper can actually run in the DE context.
@@ -235,12 +242,10 @@ def _wrapper_shebang_safe(wrapper: Path) -> bool:
interpreter itself). A python-shebang wrapper is safe only when its interpreter stays inside
the RUNNING venv; anything unknown fails safe toward the module fallback.
"""
try:
with open(wrapper, "rb") as fh:
head = fh.read(4096)
except OSError:
head = _read_head(wrapper)
if head is None:
return False
if head[:4] == b"\x7fELF" or head.startswith(b"MZ"):
if _is_native_binary(head):
return True
if not head.startswith(b"#!"):
return False
@@ -273,11 +278,10 @@ def _wrapper_targets_checkout(wrapper: Path, checkout_root: Path) -> bool:
tried: the installer writes $INSTALL_DIR lexically, so with a symlinked home the shim text
carries the lexical path while the caller may pass a resolved one.
"""
try:
head = wrapper.read_bytes()[:4096]
except OSError:
head = _read_head(wrapper)
if head is None:
return False
if b"\x7fELF" in head[:4] or head.startswith(b"MZ"):
if _is_native_binary(head):
# Native binary: cannot verify, cannot be another checkout's bash shim either — accept.
return True
text = head.decode("utf-8", errors="replace")
@@ -322,14 +326,10 @@ def _needs_interpreter(bin_path: Path) -> bool:
Native binaries (uv shim, PyInstaller, distro package) and shell wrappers (the installer's
bash launcher execs the venv python itself) never need one.
"""
try:
with open(bin_path, "rb") as fh:
head = fh.readline(256)
except OSError:
head = _read_head(bin_path, 256)
if head is None or not head.startswith(b"#!"):
return False
if not head.startswith(b"#!"):
return False
shebang = head.decode("utf-8", errors="replace").strip()
shebang = head.decode("utf-8", errors="replace").splitlines()[0].strip()
if "python" not in shebang.lower():
return False
return _shebang_escapes_running_env(shebang)
@@ -400,17 +400,19 @@ def refresh_desktop_databases(applications_dir: Path) -> "list[str]":
return ran
def _run_quiet(cmd: "list[str]") -> bool:
def _run_quiet(cmd: "list[str]", *, timeout: int = 60, on_error: Optional[bool] = False, **kwargs) -> Optional[bool]:
"""Exit-status success of a silenced subprocess; ``on_error`` when it could not be run at all."""
try:
result = subprocess.run(
cmd,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
timeout=60,
timeout=timeout,
**kwargs,
)
except (OSError, subprocess.SubprocessError):
return False
return on_error
return result.returncode == 0

46
hermes_cli/loops.py
View File

@@ -231,9 +231,9 @@ class LoopState:
"last_stop_reason": data.get("last_stop_reason"),
"route": route if isinstance(route, dict) else {},
}
# Remaining numeric/str fields: missing key -> dataclass default; present-but-falsy -> zero.
for f in fields(cls):
if f.name not in kwargs:
# Missing key -> dataclass default; present-but-falsy -> the type's zero.
cast = {"str": str, "int": int, "float": float}[f.type]
kwargs[f.name] = cast(data.get(f.name, f.default) or cast())
return cls(**kwargs)
@@ -472,35 +472,34 @@ class LoopManager:
route=dict(route or {}),
)
self._state = state
save_loop(self.session_id, state)
return state
return self._save()
def _save(self) -> LoopState:
save_loop(self.session_id, self._state)
return self._state
def pause(self, reason: str = "user-paused") -> Optional[LoopState]:
if not self._state or self._state.status not in {"active", "paused"}:
s = self._state
if not s or s.status not in {"active", "paused"}:
return None
self._state.status = "paused"
self._state.paused_reason = reason
self._state.awaiting_response = False
save_loop(self.session_id, self._state)
return self._state
s.status, s.paused_reason, s.awaiting_response = "paused", reason, False
return self._save()
def resume(self) -> Optional[LoopState]:
if not self._state or self._state.status == "cleared":
s = self._state
if not s or s.status == "cleared":
return None
self._state.status = "active"
self._state.paused_reason = None
self._state.awaiting_response = False
s.status, s.paused_reason, s.awaiting_response = "active", None, False
# Re-arm relative to now so a long pause doesn't fire instantly N times.
delay = self._state.current_delay or self._state.interval_seconds or self_paced_floor_seconds()
self._state.next_due_at = time.time() + min(delay, 5.0)
save_loop(self.session_id, self._state)
return self._state
delay = s.current_delay or s.interval_seconds or self_paced_floor_seconds()
s.next_due_at = time.time() + min(delay, 5.0)
return self._save()
def clear(self) -> bool:
if self._state is None or self._state.status == "cleared":
return False
self._state.status = "cleared"
save_loop(self.session_id, self._state)
self._save()
self._state = None
return True
@@ -526,9 +525,8 @@ class LoopManager:
s.awaiting_response = True
# Provisional schedule from NOW: complete_tick reschedules from turn end, but if the
# process dies mid-turn this keeps the persisted loop from being 'due' in a tight loop.
delay = s.current_delay or s.interval_seconds or self_paced_floor_seconds()
s.next_due_at = s.last_fired_at + delay
save_loop(self.session_id, s)
s.next_due_at = s.last_fired_at + (s.current_delay or s.interval_seconds or self_paced_floor_seconds())
self._save()
if s.prompt.lstrip().startswith("/"):
return s.prompt.strip()
@@ -543,7 +541,7 @@ class LoopManager:
return
s.awaiting_response = False
s.ticks_fired = max(0, s.ticks_fired - 1)
save_loop(self.session_id, s)
self._save()
def _stop(self, status: str, reason: str, message: str) -> Dict[str, Any]:
"""Persist a terminal (``done``) or recoverable (``paused``) stop and build the result."""
@@ -553,7 +551,7 @@ class LoopManager:
s.last_stop_reason = reason
else:
s.paused_reason = reason
save_loop(self.session_id, s)
self._save()
return {"status": status, "stopped": True, "reason": reason, "message": message}
def complete_tick(self, last_response: str) -> Dict[str, Any]:
@@ -616,7 +614,7 @@ class LoopManager:
else:
s.current_delay = s.interval_seconds
s.next_due_at = now + s.current_delay
save_loop(self.session_id, s)
self._save()
return {"status": "active", "stopped": False, "reason": "loop continues", "message": ""}