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

refactor(hermes_cli): group D — drop blank line after local imports (AST-neutral)

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:10:21 -07:00
parent 2295bcc15a
commit 2cac493fec
5 changed files with 0 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
hermes_cli/container_boot.py
View File

@@ -87,7 +87,6 @@ def reconcile_profile_gateways(
# precedence over config.yaml and otherwise preserves the configured value.
from gateway.config import load_gateway_config
from utils import is_truthy_value
try:
multiplex_profiles = load_gateway_config().multiplex_profiles
except Exception:

16
hermes_cli/cron.py
View File

@@ -35,7 +35,6 @@ def _normalize_skills(single_skill=None, skills: Optional[Iterable[str]] = None)
def _cron_api(**kwargs):
from tools.cronjob_tools import cronjob as cronjob_tool
return json.loads(cronjob_tool(**kwargs))
@@ -47,7 +46,6 @@ def _active_cron_provider_name() -> str:
"""
try:
from cron.scheduler_provider import resolve_cron_scheduler
return resolve_cron_scheduler().name or "builtin"
except Exception:
return "builtin"
@@ -70,7 +68,6 @@ def _builtin_gateway_liveness() -> Optional[bool]:
# gateway just after a restart).
try:
from gateway.status import is_gateway_runtime_lock_active
if is_gateway_runtime_lock_active():
return True
except Exception:
@@ -162,7 +159,6 @@ _STATE_BADGES = {"paused": ("[paused]", Colors.YELLOW), "completed": ("[complete
def cron_list(show_all: bool = False):
"""List all scheduled jobs."""
from cron.jobs import effective_job_state, list_jobs
jobs = list_jobs(include_disabled=show_all)
if not jobs:
@@ -289,7 +285,6 @@ def cron_tick():
def cron_runs(job_id: Optional[str] = None, limit: int = 20):
"""Show indexed durable cron execution history."""
from cron.executions import list_executions
records = list_executions(job_id=job_id, limit=limit)
if not records:
print("No cron execution attempts recorded.")
@@ -312,7 +307,6 @@ def cron_incidents(args) -> int:
acking closes an incident so its failure ping stays silent until the error signature changes.
"""
from cron.incidents import ack_incident, list_incidents
action = getattr(args, "incident_action", "list")
if action == "ack":
incident_id = getattr(args, "incident_id", None)
@@ -372,7 +366,6 @@ def _print_ticker_health(pids: list) -> None:
TICKER_INTERVAL_SECONDS,
)
from cron.scheduler import _is_fd_exhaustion_text as _cron_is_fd_exhaustion_text
# ~3 missed ticker iterations (+ slack) before declaring trouble; derived from the shared
# interval so the threshold tracks the ticker cadence (= 200s at the 60s default).
STALE_AFTER = TICKER_INTERVAL_SECONDS * 3 + 20
@@ -423,7 +416,6 @@ def cron_status():
"""Show cron execution status."""
from cron.jobs import list_jobs
from hermes_cli.gateway import find_gateway_pids
print()
provider = _active_cron_provider_name()
@@ -444,7 +436,6 @@ def cron_status():
# process is alive. Only declare "not running" when both agree.
try:
from gateway.status import get_running_pid, is_gateway_runtime_lock_active
gateway_alive_via_lock = is_gateway_runtime_lock_active()
lock_pid = get_running_pid() if gateway_alive_via_lock else None
pids = [lock_pid] if lock_pid else pids
@@ -500,7 +491,6 @@ def _scripts_dir_for_cron() -> Path:
profile-aware callers that monkeypatch cron storage inspect the same Hermes home.
"""
from cron.jobs import CRON_DIR
return CRON_DIR.parent / "scripts"
@@ -531,7 +521,6 @@ _OVERDUE_GRACE_SECONDS = 15 * 60
def _next_run_overdue_issue(next_run: str) -> Optional[str]:
"""Issue string when ``next_run_at`` is parked in the past."""
from datetime import datetime, timezone
try:
dt = datetime.fromisoformat(next_run.replace("Z", "+00:00"))
except ValueError:
@@ -586,7 +575,6 @@ def _cron_doctor_issues_for_job(job: Dict[str, Any]) -> List[str]:
def cron_doctor() -> int:
"""Run read-only cron health checks and return a shell-friendly status."""
from cron.jobs import list_jobs
jobs = list_jobs(include_disabled=False)
findings = [(job, issues) for job in jobs if (issues := _cron_doctor_issues_for_job(job))]
@@ -662,7 +650,6 @@ def cron_create(args):
def cron_edit(args):
from cron.jobs import AmbiguousJobReference, resolve_job_ref
try:
job = resolve_job_ref(args.job_id)
except AmbiguousJobReference as exc:
@@ -716,7 +703,6 @@ def _job_action(action: str, job_id: str, success_verb: str) -> int:
# callers (tests, embedding apps) are not tainted.
try:
from gateway.session_context import _SESSION_ASYNC_DELIVERY
_stateless_token = _SESSION_ASYNC_DELIVERY.set(False)
except Exception:
_stateless_token = None
@@ -761,7 +747,6 @@ def cron_resume(args) -> int:
return 1
return _job_action("resume", args.job_id, "Resumed")
from cron.jobs import AmbiguousJobReference, _hermes_now, rearm_oneshot
if run_now:
run_at = _hermes_now().isoformat()
try:
@@ -785,7 +770,6 @@ def cron_notepad(args) -> int:
notepads into the job prompt on each run.
"""
from cron import notepad
job_id = str(getattr(args, "job_id", "") or "")
action = getattr(args, "notepad_action", None) or "list"
key = getattr(args, "key", None)

10
hermes_cli/dashboard_procs.py
View File

@@ -23,7 +23,6 @@ _PS_RUN_KWARGS = dict(capture_output=True, text=True, encoding="utf-8", errors="
def _m():
"""Lazy ``hermes_cli.main`` reference (call-time; keeps patches working)."""
from hermes_cli import main
return main
@@ -47,7 +46,6 @@ def _iter_process_table() -> list[tuple[int, str]]:
# readers unbounded and a conhost descendant holding duplicated handles wedges it
# forever. It also passes CREATE_NO_WINDOW for the pythonw.exe backend.
from hermes_cli._subprocess_compat import bounded_probe_run
result = bounded_probe_run(
["wmic", "process", "get", "ProcessId,CommandLine", "/FORMAT:LIST"],
timeout=10, errors="ignore",
@@ -98,7 +96,6 @@ def _scan_dashboard_processes(*, exclude_pids: set[int] | None = None) -> list[t
# missed, preferring the ledger's full argv.
try:
from hermes_cli.process_identity import ledger_entries
seen = {pid for pid, _ in found} | skip
for entry in ledger_entries():
pid = entry.get("pid")
@@ -116,7 +113,6 @@ def _hermes_home_for_pid(pid: int) -> str | None:
"""Best-effort ``HERMES_HOME`` from *pid*'s environment (psutil, then /proc)."""
try:
import psutil
home = psutil.Process(pid).environ().get("HERMES_HOME")
if home:
return home
@@ -234,7 +230,6 @@ def _filter_dashboard_respawn_candidates(
if own_home is None:
try:
from hermes_constants import get_hermes_home
own_home = str(get_hermes_home())
except Exception:
own_home = ""
@@ -278,7 +273,6 @@ def _kill_pids_windows(pids: list[int], killed: list[int], failed: list[tuple[in
"""``taskkill /F`` each PID after re-verifying its identity."""
from gateway.status import get_process_start_time
from hermes_cli._subprocess_compat import pid_is_hermes, windows_hide_flags
# Capture identity immediately after discovery: a PID reused before the destructive
# action fails the start-time check.
pid_start_times = {pid: get_process_start_time(pid) for pid in pids}
@@ -610,7 +604,6 @@ def _lock_owned_serve_pids(base_dir: Path | None = None) -> set[int]:
Best-effort: a bad record contributes no PID; never raises.
"""
import json
root = base_dir if base_dir is not None else _hermes_home_dir() / _REMOTE_LOCK_SUBDIR
owned: set[int] = set()
if not root.is_dir():
@@ -647,7 +640,6 @@ def _process_age_seconds(pid: int) -> float:
import time as _time
import psutil as _psutil
return max(0.0, _time.time() - _psutil.Process(pid).create_time())
@@ -670,7 +662,6 @@ def _reap_orphaned_desktop_local_serves(
"""
import signal as _signal
import time as _time
signal_term = _signal.SIGTERM if signal_term is None else signal_term
signal_kill = getattr(_signal, "SIGKILL", _signal.SIGTERM) if signal_kill is None else signal_kill
sleep_fn = sleep_fn or _time.sleep
@@ -733,7 +724,6 @@ def _reap_orphaned_desktop_local_serves(
# which is a Windows footgun the linter blocks everywhere.
sleep_fn(1.5)
import psutil
for pid in matched:
if pid in failed:
continue

5
hermes_cli/dashboard_register.py
View File

@@ -52,7 +52,6 @@ def _resolve_portal_base_url(override: Optional[str] = None) -> str:
return override.rstrip("/")
try:
from hermes_cli.auth import DEFAULT_NOUS_PORTAL_URL, get_provider_auth_state
base = (get_provider_auth_state("nous") or {}).get("portal_base_url")
chosen = base if isinstance(base, str) and base.strip() else str(DEFAULT_NOUS_PORTAL_URL)
return chosen.rstrip("/")
@@ -113,7 +112,6 @@ def _print_post_register_hint(
) -> None:
"""Print the success summary + the gate-engagement caveat."""
from hermes_cli.config import get_env_path
print(f"\n Wrote to {get_env_path()}:\n HERMES_DASHBOARD_OAUTH_CLIENT_ID={client_id}")
if wrote_portal_url:
print(" HERMES_DASHBOARD_PORTAL_URL=" + str(portal_base_url))
@@ -150,7 +148,6 @@ def _print_post_register_hint(
def _env_value(key: str) -> Optional[str]:
"""Stored ``.env`` value, or ``None`` on any read failure."""
from hermes_cli.config import get_env_value
try:
return get_env_value(key)
except Exception:
@@ -160,7 +157,6 @@ def _env_value(key: str) -> Optional[str]:
def _save_env_quietly(key: str, value: str) -> bool:
"""Persist *key*; False on failure (non-fatal: only client_id is load-bearing)."""
from hermes_cli.config import save_env_value
try:
save_env_value(key, value)
return True
@@ -188,7 +184,6 @@ def cmd_dashboard_register(args) -> None:
"""Register a self-hosted dashboard OAuth client with Nous Portal."""
from hermes_cli.auth import AuthError, resolve_nous_access_token
from hermes_cli.config import is_managed, save_env_value
# Managed (Docker/hosted) installs get HERMES_DASHBOARD_OAUTH_CLIENT_ID stamped in by the
# orchestrator; save_env_value refuses to write anyway.
if is_managed():

4
hermes_cli/debug.py
View File

@@ -242,7 +242,6 @@ class LogSnapshot:
def _primary_log_path(log_name: str) -> Optional[Path]:
"""Where *log_name* would live if present. Doesn't check existence."""
from hermes_cli.logs import LOG_FILES
filename = LOG_FILES.get(log_name)
return (get_hermes_home() / "logs" / filename) if filename else None
@@ -290,7 +289,6 @@ def _redact_log_text(text: str) -> str:
if not text:
return text
from agent.redact import redact_sensitive_text
text = redact_sensitive_text(text, force=True)
return _EMAIL_ADDRESS_RE.sub("[REDACTED_EMAIL]", text)
@@ -436,7 +434,6 @@ def collect_debug_report(
# process, where the errors.log tail below carries the same escalation lines instead.
try:
from agent.agent_runtime_helpers import get_sanitizer_heal_stats
heal_stats = get_sanitizer_heal_stats()
if heal_stats:
buf.write("\n\n--- transcript sanitiser heal counters ---\n")
@@ -638,7 +635,6 @@ _NOUS_PRIVACY_NOTICE = """\
def _run_debug_share_nous(args, *, log_lines: int, redact: bool) -> None:
"""``hermes debug share --nous``: gzip the same bundle into the Nous envelope → Nous-S3."""
from hermes_cli.diagnostics_upload import share_to_nous
print(_NOUS_PRIVACY_NOTICE)
if not _confirm_upload(args):
return