refactor: remove redundant local imports already available at module level

Sweep ~74 redundant local imports across 21 files where the same module
was already imported at the top level. Also includes type fixes and lint
cleanups on the same branch.
This commit is contained in:
alt-glitch 2026-04-21 12:35:10 +05:30 committed by Teknium
parent ce9c91c8f7
commit 1010e5fa3c
31 changed files with 289 additions and 316 deletions

View file

@ -1911,7 +1911,6 @@ def _maybe_start_recording(task_id: str):
recordings_dir.mkdir(parents=True, exist_ok=True)
_cleanup_old_recordings(max_age_hours=72)
import time
timestamp = time.strftime("%Y%m%d_%H%M%S")
recording_path = recordings_dir / f"session_{timestamp}_{task_id[:16]}.webm"
@ -2027,8 +2026,6 @@ def browser_vision(question: str, annotate: bool = False, task_id: Optional[str]
import base64
import uuid as uuid_mod
from pathlib import Path
effective_task_id = task_id or "default"
# Save screenshot to persistent location so it can be shared with users
@ -2210,7 +2207,6 @@ def _cleanup_old_screenshots(screenshots_dir, max_age_hours=24):
def _cleanup_old_recordings(max_age_hours=72):
"""Remove browser recordings older than max_age_hours to prevent disk bloat."""
import time
try:
hermes_home = get_hermes_home()
recordings_dir = hermes_home / "browser_recordings"

View file

@ -389,7 +389,6 @@ class CheckpointManager:
@staticmethod
def _parse_shortstat(stat_line: str, entry: Dict) -> None:
"""Parse git --shortstat output into entry dict."""
import re
m = re.search(r'(\d+) file', stat_line)
if m:
entry["files_changed"] = int(m.group(1))

View file

@ -1540,7 +1540,6 @@ def _interrupted_call_result() -> str:
def _interpolate_env_vars(value):
"""Recursively resolve ``${VAR}`` placeholders from ``os.environ``."""
if isinstance(value, str):
import re
def _replace(m):
return os.environ.get(m.group(1), m.group(0))
return re.sub(r"\$\{([^}]+)\}", _replace, value)

View file

@ -1167,32 +1167,31 @@ PROCESS_SCHEMA = {
def _handle_process(args, **kw):
import json as _json
task_id = kw.get("task_id")
action = args.get("action", "")
# Coerce to string — some models send session_id as an integer
session_id = str(args.get("session_id", "")) if args.get("session_id") is not None else ""
if action == "list":
return _json.dumps({"processes": process_registry.list_sessions(task_id=task_id)}, ensure_ascii=False)
return json.dumps({"processes": process_registry.list_sessions(task_id=task_id)}, ensure_ascii=False)
elif action in ("poll", "log", "wait", "kill", "write", "submit", "close"):
if not session_id:
return tool_error(f"session_id is required for {action}")
if action == "poll":
return _json.dumps(process_registry.poll(session_id), ensure_ascii=False)
return json.dumps(process_registry.poll(session_id), ensure_ascii=False)
elif action == "log":
return _json.dumps(process_registry.read_log(
return json.dumps(process_registry.read_log(
session_id, offset=args.get("offset", 0), limit=args.get("limit", 200)), ensure_ascii=False)
elif action == "wait":
return _json.dumps(process_registry.wait(session_id, timeout=args.get("timeout")), ensure_ascii=False)
return json.dumps(process_registry.wait(session_id, timeout=args.get("timeout")), ensure_ascii=False)
elif action == "kill":
return _json.dumps(process_registry.kill_process(session_id), ensure_ascii=False)
return json.dumps(process_registry.kill_process(session_id), ensure_ascii=False)
elif action == "write":
return _json.dumps(process_registry.write_stdin(session_id, str(args.get("data", ""))), ensure_ascii=False)
return json.dumps(process_registry.write_stdin(session_id, str(args.get("data", ""))), ensure_ascii=False)
elif action == "submit":
return _json.dumps(process_registry.submit_stdin(session_id, str(args.get("data", ""))), ensure_ascii=False)
return json.dumps(process_registry.submit_stdin(session_id, str(args.get("data", ""))), ensure_ascii=False)
elif action == "close":
return _json.dumps(process_registry.close_stdin(session_id), ensure_ascii=False)
return json.dumps(process_registry.close_stdin(session_id), ensure_ascii=False)
return tool_error(f"Unknown process action: {action}. Use: list, poll, log, wait, kill, write, submit, close")

View file

@ -509,7 +509,6 @@ def _get_disabled_skill_names() -> Set[str]:
def _is_skill_disabled(name: str, platform: str = None) -> bool:
"""Check if a skill is disabled in config."""
import os
try:
from hermes_cli.config import load_config
config = load_config()

View file

@ -217,7 +217,6 @@ def _prompt_for_sudo_password(timeout_seconds: int = 45) -> str:
directly from /dev/tty with echo disabled.
"""
import sys
import time as time_module
# Use the registered callback when available (prompt_toolkit-compatible)
if _sudo_password_callback is not None:
@ -278,7 +277,7 @@ def _prompt_for_sudo_password(timeout_seconds: int = 45) -> str:
try:
os.environ["HERMES_SPINNER_PAUSE"] = "1"
time_module.sleep(0.2)
time.sleep(0.2)
print()
print("" + "" * 58 + "")