diff --git a/hermes_cli/container_boot.py b/hermes_cli/container_boot.py index 5266f363f5..262b0d838a 100644 --- a/hermes_cli/container_boot.py +++ b/hermes_cli/container_boot.py @@ -275,7 +275,7 @@ def _cleanup_stale_runtime_files(profile_dir: Path) -> None: def _read_prior_exit_label(profile_dir: Path) -> str: - """Exception-free ``lifecycle_ledger.read_prior_exit_label``: forensics never block cont-init.""" + """Exception-free ``lifecycle_ledger.read_prior_exit_label`` — forensics never block boot.""" try: from gateway.lifecycle_ledger import read_prior_exit_label return read_prior_exit_label(profile_dir) diff --git a/hermes_cli/cron.py b/hermes_cli/cron.py index 7bcc5bdb31..9d67b08b68 100644 --- a/hermes_cli/cron.py +++ b/hermes_cli/cron.py @@ -21,13 +21,10 @@ from cron.lifecycle_guard import ( # noqa: F401 (re-exported for terminal_tool def _normalize_skills(single_skill=None, skills: Optional[Iterable[str]] = None) -> Optional[List[str]]: - if skills is None: - if single_skill is None: - return None - raw_items = [single_skill] - else: - raw_items = list(skills) - + """Deduped, stripped skill names; None when neither argument was given.""" + if skills is None and single_skill is None: + return None + raw_items = list(skills) if skills is not None else [single_skill] normalized: List[str] = [] for item in raw_items: text = str(item or "").strip() @@ -78,7 +75,9 @@ def _builtin_gateway_liveness() -> Optional[bool]: return True except Exception: pass # a crashing lock probe is "unknown", not "dead" — let the pid scan decide - from hermes_cli.gateway import find_gateway_pids, named_profile_served_by_running_multiplexer + from hermes_cli.gateway import ( + find_gateway_pids, named_profile_served_by_running_multiplexer, + ) if find_gateway_pids(): return True @@ -161,7 +160,9 @@ def _print_banner(title: str) -> None: def _unverified_targets(unverified) -> str: - return ", ".join(str(t) for t in unverified) if isinstance(unverified, list) else str(unverified) + if isinstance(unverified, list): + return ", ".join(str(t) for t in unverified) + return str(unverified) _STATE_BADGES = {"paused": ("[paused]", Colors.YELLOW), "completed": ("[completed]", Colors.BLUE)} @@ -217,7 +218,8 @@ def cron_list(show_all: bool = False): if mon_state.get("last_changed_at"): rows.append(("Changed", mon_state["last_changed_at"])) if job.get("no_agent"): - rows.append(("Mode", f"{color('no-agent', Colors.DIM)} (script stdout delivered directly)")) + mode = color("no-agent", Colors.DIM) + " (script stdout delivered directly)" + rows.append(("Mode", mode)) if job.get("workdir"): rows.append(("Workdir", job["workdir"])) @@ -410,7 +412,7 @@ def _print_ticker_health(pids: list) -> None: ) print(" Cron jobs may NOT be firing. Restart: hermes gateway restart") elif ok_age is not None and ok_age > STALE_AFTER: - # Loop alive (fresh heartbeat) but no tick SUCCEEDED in a long time → failing every iteration. + # Loop alive (fresh heartbeat) but no tick SUCCEEDED in a long time → every tick fails. _warn( "⚠ Gateway and cron ticker are running, but no tick has " f"succeeded in {int(ok_age)}s — ticks may be failing." @@ -503,7 +505,8 @@ def _print_active_jobs_summary(jobs) -> None: ] if late: print() - print(color(f" ⚠ {len(late)} job(s) last fired late (missed-fire catch-up):", Colors.YELLOW)) + print(color(f" ⚠ {len(late)} job(s) last fired late (missed-fire catch-up):", + Colors.YELLOW)) for j in late: d = j["last_dispatch"] print( @@ -584,26 +587,20 @@ def _cron_doctor_issues_for_job(job: Dict[str, Any]) -> List[str]: unverified = job.get("last_delivery_unverified") if unverified: - issues.append( - f"last delivery unverified (adapter acked without evidence): {_unverified_targets(unverified)}" - ) + issues.append("last delivery unverified (adapter acked without evidence): " + + _unverified_targets(unverified)) if job.get("enabled", True) and job.get("state") not in {"paused", "completed"}: next_run = str(job.get("next_run_at") or "").strip() - if not next_run: - issues.append("active job has no next_run_at") - else: - overdue = _next_run_overdue_issue(next_run) - if overdue: - issues.append(overdue) + issue = _next_run_overdue_issue(next_run) if next_run else "active job has no next_run_at" + if issue: + issues.append(issue) script = str(job.get("script") or "").strip() if job.get("no_agent") and not script: issues.append("no-agent job has no script") - if script: - script_issue = _script_health_issue(script) - if script_issue: - issues.append(script_issue) + if script and (script_issue := _script_health_issue(script)): + issues.append(script_issue) workdir = str(job.get("workdir") or "").strip() if workdir and not Path(workdir).expanduser().exists(): @@ -617,11 +614,7 @@ def cron_doctor() -> int: from cron.jobs import list_jobs jobs = list_jobs(include_disabled=False) - findings: List[tuple[Dict[str, Any], List[str]]] = [] - for job in jobs: - issues = _cron_doctor_issues_for_job(job) - if issues: - findings.append((job, issues)) + findings = [(job, issues) for job in jobs if (issues := _cron_doctor_issues_for_job(job))] if not findings: print(color("✓ Cron doctor found no issues", Colors.GREEN)) @@ -635,9 +628,7 @@ def cron_doctor() -> int: print(color(f"Cron doctor found {issue_count} issue(s) across {len(findings)} job(s):", Colors.YELLOW)) print() for job, issues in findings: - job_id = job.get("id", "?") - name = job.get("name", "(unnamed)") - print(f" {color(job_id, Colors.YELLOW)} {name}") + print(f" {color(job.get('id', '?'), Colors.YELLOW)} {job.get('name', '(unnamed)')}") for issue in issues: print(f" - {issue}") print() @@ -724,9 +715,7 @@ def cron_edit(args): final_skills = replacement_skills elif add_skills or remove_skills: final_skills = [skill for skill in existing_skills if skill not in remove_skills] - for skill in add_skills: - if skill not in final_skills: - final_skills.append(skill) + final_skills += [skill for skill in add_skills if skill not in final_skills] result = _cron_api(action="update", job_id=args.job_id, schedule=getattr(args, "schedule", None), @@ -782,14 +771,12 @@ def _job_action(action: str, job_id: str, success_verb: str) -> int: # (execution_mode="background" and/or a delegation_id) and keeps running AFTER this # CLI exits — a terminal success/failure verdict would be a lie, so report the dispatch. delegation_id = job.get("delegation_id") - if job.get("execution_mode") == "background" or delegation_id: - if delegation_id: - print(f" Running in background (delegation {delegation_id}).") - else: - print(" Running in background.") + if delegation_id: + print(f" Running in background (delegation {delegation_id}).") + elif job.get("execution_mode") == "background": + print(" Running in background.") elif job.get("executed"): - outcome = "succeeded" if job.get("execution_success") else "failed" - print(f" Ran now: {outcome}.") + print(f" Ran now: {'succeeded' if job.get('execution_success') else 'failed'}.") elif job.get("execution_skipped"): print(f" {job['execution_skipped']}") else: @@ -862,7 +849,6 @@ def cron_notepad(args) -> int: print(color(f"No notepad key '{key}' for job {job_id}.", Colors.YELLOW)) return 1 - # list (default) notes = notepad.list_notes(job_id) if not notes: print(color(f"Notepad for job {job_id} is empty.", Colors.DIM)) diff --git a/hermes_cli/dashboard_procs.py b/hermes_cli/dashboard_procs.py index 8e672c0f04..0171877f09 100644 --- a/hermes_cli/dashboard_procs.py +++ b/hermes_cli/dashboard_procs.py @@ -102,7 +102,9 @@ def _scan_dashboard_processes(*, exclude_pids: set[int] | None = None) -> list[t seen = {pid for pid, _ in found} | skip for entry in ledger_entries(): pid = entry.get("pid") - if entry.get("purpose") in ("serve", "dashboard") and isinstance(pid, int) and pid not in seen: + if entry.get("purpose") not in ("serve", "dashboard") or not isinstance(pid, int): + continue + if pid not in seen: found.append((pid, str(entry.get("argv") or ""))) except Exception: pass # ledger unavailable → scan-only behavior @@ -540,7 +542,7 @@ def _is_desktop_local_serve_cmdline(command: str) -> bool: def _process_ppid(pid: int) -> int | None: - """Best-effort parent pid; None on failure (always on Windows: desktop tree-kill reaps there).""" + """Best-effort parent pid; None on failure (always None on Windows: desktop tree-kill reaps).""" try: if sys.platform == "win32": return None diff --git a/hermes_cli/dashboard_register.py b/hermes_cli/dashboard_register.py index 0dca8de774..03751e916a 100644 --- a/hermes_cli/dashboard_register.py +++ b/hermes_cli/dashboard_register.py @@ -99,7 +99,8 @@ def _register_self_hosted_client( raise RuntimeError( detail or "Your account is not permitted to register a self-hosted dashboard." ) from exc - raise RuntimeError(f"Portal returned HTTP {exc.code}" + (f": {detail}" if detail else "")) from exc + suffix = f": {detail}" if detail else "" + raise RuntimeError(f"Portal returned HTTP {exc.code}{suffix}") from exc except urllib.error.URLError as exc: raise RuntimeError(f"Could not reach Nous Portal at {portal_base_url}: {exc.reason}") from exc diff --git a/hermes_cli/debug.py b/hermes_cli/debug.py index 3403f33c42..9cfc106507 100644 --- a/hermes_cli/debug.py +++ b/hermes_cli/debug.py @@ -63,7 +63,7 @@ def _load_pending() -> list[dict]: return [] if not isinstance(data, list): return [] - return [e for e in data if isinstance(e, dict) and "url" in e and "expire_at" in e] + return [e for e in data if isinstance(e, dict) and {"url", "expire_at"} <= e.keys()] def _save_pending(entries: list[dict]) -> None: @@ -99,18 +99,14 @@ def _sweep_expired_pastes(now: Optional[float] = None) -> tuple[int, int]: if expire_at > current: remaining.append(entry) continue - try: - if delete_paste(entry.get("url", "")): - deleted += 1 - continue + gone = delete_paste(entry.get("url", "")) except Exception: - pass # network hiccup, 404 (already gone), ... - - if expire_at + 86400 > current: - remaining.append(entry) + gone = False # network hiccup, 404 (already gone), ... + if gone or expire_at + 86400 <= current: + deleted += 1 # deleted, or given up on → count as reaped else: - deleted += 1 # count as reaped + remaining.append(entry) if deleted: _save_pending(remaining) @@ -299,7 +295,9 @@ def _redact_log_text(text: str) -> str: return _EMAIL_ADDRESS_RE.sub("[REDACTED_EMAIL]", text) -def _read_tail_bytes(log_path: Path, size: int, max_bytes: int, tail_lines: int) -> tuple[bytes, bool]: +def _read_tail_bytes( + log_path: Path, size: int, max_bytes: int, tail_lines: int, +) -> tuple[bytes, bool]: """Read the whole file, or enough of its tail for both views → (raw, truncated). For oversized files, read backwards until we have ``max_bytes`` for the standalone upload @@ -313,7 +311,8 @@ def _read_tail_bytes(log_path: Path, size: int, max_bytes: int, tail_lines: int) chunks: list[bytes] = [] total = 0 newline_count = 0 - while pos > 0 and (total < max_bytes or newline_count <= tail_lines + 1) and total < max_bytes * 2: + while (pos > 0 and total < max_bytes * 2 + and (total < max_bytes or newline_count <= tail_lines + 1)): read_size = min(chunk_size, pos) pos -= read_size f.seek(pos) @@ -384,7 +383,9 @@ def _tail_budget(name: str, log_lines: int) -> int: return log_lines if name == "agent" else min(log_lines, 100) -def _capture_default_log_snapshots(log_lines: int, *, redact: bool = True) -> dict[str, LogSnapshot]: +def _capture_default_log_snapshots( + log_lines: int, *, redact: bool = True, +) -> dict[str, LogSnapshot]: """Capture all logs used by debug-share exactly once.""" return { name: _capture_log_snapshot(name, tail_lines=_tail_budget(name, log_lines), redact=redact) @@ -414,7 +415,8 @@ def _capture_dump() -> str: def collect_debug_report( - *, log_lines: int = 200, dump_text: str = "", log_snapshots: Optional[dict[str, LogSnapshot]] = None, + *, log_lines: int = 200, dump_text: str = "", + log_snapshots: Optional[dict[str, LogSnapshot]] = None, ) -> str: """Build the summary debug report (system dump + log tails) as upload-ready text. @@ -471,7 +473,8 @@ def collect_share_bundle(log_lines: int = 200, redact: bool = True) -> dict[str, dump_text = _capture_dump() log_snapshots = _capture_default_log_snapshots(log_lines, redact=redact) - report = collect_debug_report(log_lines=log_lines, dump_text=dump_text, log_snapshots=log_snapshots) + report = collect_debug_report(log_lines=log_lines, dump_text=dump_text, + log_snapshots=log_snapshots) banner = _REDACTION_BANNER if redact else "" bundle: dict[str, str] = {"report": banner + report} for name in _FULL_LOGS: @@ -506,7 +509,9 @@ class DebugShareResult: report: str = "" # the summary report text (kept for local fallback) -def build_debug_share(*, log_lines: int = 200, expiry: int = 7, redact: bool = True) -> DebugShareResult: +def build_debug_share( + *, log_lines: int = 200, expiry: int = 7, redact: bool = True, +) -> DebugShareResult: """Collect the debug report + full logs, upload each, return the URLs. Shared core behind ``hermes debug share`` and the dashboard ``POST /api/ops/debug-share``. @@ -517,7 +522,9 @@ def build_debug_share(*, log_lines: int = 200, expiry: int = 7, redact: bool = T bundle = collect_share_bundle(log_lines=log_lines, redact=redact) if redact: - logger.info("hermes debug share: applied force-mode redaction to log snapshots before upload") + logger.info( + "hermes debug share: applied force-mode redaction to log snapshots before upload" + ) report = bundle["report"] @@ -607,12 +614,9 @@ def run_debug_share(args): print("\nDebug report uploaded:") for label, url in result.urls.items(): print(f" {label:<{label_width}} {url}") - if result.failures: print(f"\n (failed to upload: {', '.join(result.failures)})") - - hours = result.auto_delete_seconds // 3600 - print(f"\n⏱ Pastes will auto-delete in {hours} hours.") + print(f"\n⏱ Pastes will auto-delete in {result.auto_delete_seconds // 3600} hours.") print("To delete now: hermes debug delete ") print("\nShare these links with the Hermes team for support.") @@ -633,7 +637,7 @@ _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, upload to Nous-S3.""" + """``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) @@ -664,17 +668,12 @@ def _run_debug_share_nous(args, *, log_lines: int, redact: bool) -> None: sys.exit(1) view_url = res.get("viewUrl") or res.get("view_url") - print("\nDebug bundle uploaded to Nous (private):") - if view_url: - print(f" View URL {view_url}") - else: - print(f" (no view URL returned; upload id: {res.get('id', '?')})") - expires_at = res.get("expiresAt") or res.get("expires_at") - if expires_at: - print(f"\n⏱ Auto-deletes at {expires_at} (14-day retention).") - else: - print("\n⏱ Auto-deletes after 14 days.") + print("\nDebug bundle uploaded to Nous (private):") + print(f" View URL {view_url}" if view_url + else f" (no view URL returned; upload id: {res.get('id', '?')})") + print(f"\n⏱ Auto-deletes at {expires_at} (14-day retention)." if expires_at + else "\n⏱ Auto-deletes after 14 days.") print( "\nShare this private link with the Nous team — only Nous staff " @@ -713,13 +712,12 @@ def run_debug(args): # Opportunistic sweep of expired pastes on every ``hermes debug`` call. _best_effort_sweep_expired_pastes() - subcmd = getattr(args, "debug_command", None) - if subcmd == "share": - run_debug_share(args) - elif subcmd == "delete": - run_debug_delete(args) - else: + handlers = {"share": run_debug_share, "delete": run_debug_delete} + handler = handlers.get(getattr(args, "debug_command", None)) + if handler is None: print(_DEBUG_USAGE) + else: + handler(args) _DEBUG_USAGE = """\