diff --git a/hermes_cli/linux_desktop_entry.py b/hermes_cli/linux_desktop_entry.py index 1b3665076e..a81900a4c9 100644 --- a/hermes_cli/linux_desktop_entry.py +++ b/hermes_cli/linux_desktop_entry.py @@ -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 diff --git a/hermes_cli/loops.py b/hermes_cli/loops.py index 15ea2bb62f..1d0d959e48 100644 --- a/hermes_cli/loops.py +++ b/hermes_cli/loops.py @@ -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": ""}