refactor: migrate 10 config.yaml inline loaders to read_raw_config()
Replace 10 callsites across 6 files that manually opened config.yaml,
called yaml.safe_load(), and handled missing-file/parse-error fallbacks
with the new read_raw_config() helper from hermes_cli/config.py.
Each migrated site previously had 5-8 lines of boilerplate:
config_path = get_hermes_home() / 'config.yaml'
if config_path.exists():
import yaml
with open(config_path) as f:
cfg = yaml.safe_load(f) or {}
Now reduced to:
from hermes_cli.config import read_raw_config
cfg = read_raw_config()
Migrated files:
- tools/browser_tool.py (4 sites): command_timeout, cloud_provider,
allow_private_urls, record_sessions
- tools/env_passthrough.py: terminal.env_passthrough
- tools/credential_files.py: terminal.credential_files
- tools/transcription_tools.py: stt.model
- hermes_cli/commands.py: config-gated command resolution
- hermes_cli/auth.py (2 sites): model config read + provider reset
Skipped (intentionally):
- gateway/run.py: 10+ sites with local aliases, critical path
- hermes_cli/profiles.py: profile-specific config path
- hermes_cli/doctor.py: reads raw then writes fixes back
- agent/model_metadata.py: different file (context_length_cache.yaml)
- tools/website_policy.py: custom config_path param + error types
This commit is contained in:
@@ -37,7 +37,7 @@ from typing import Any, Dict, List, Optional
|
||||
import httpx
|
||||
import yaml
|
||||
|
||||
from hermes_cli.config import get_hermes_home, get_config_path
|
||||
from hermes_cli.config import get_hermes_home, get_config_path, read_raw_config
|
||||
from hermes_constants import OPENROUTER_BASE_URL
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -2214,14 +2214,7 @@ def _update_config_for_provider(
|
||||
config_path = get_config_path()
|
||||
config_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
config: Dict[str, Any] = {}
|
||||
if config_path.exists():
|
||||
try:
|
||||
loaded = yaml.safe_load(config_path.read_text()) or {}
|
||||
if isinstance(loaded, dict):
|
||||
config = loaded
|
||||
except Exception:
|
||||
config = {}
|
||||
config = read_raw_config()
|
||||
|
||||
current_model = config.get("model")
|
||||
if isinstance(current_model, dict):
|
||||
@@ -2258,12 +2251,8 @@ def _reset_config_provider() -> Path:
|
||||
if not config_path.exists():
|
||||
return config_path
|
||||
|
||||
try:
|
||||
config = yaml.safe_load(config_path.read_text()) or {}
|
||||
except Exception:
|
||||
return config_path
|
||||
|
||||
if not isinstance(config, dict):
|
||||
config = read_raw_config()
|
||||
if not config:
|
||||
return config_path
|
||||
|
||||
model = config.get("model")
|
||||
|
||||
@@ -293,14 +293,8 @@ def _resolve_config_gates() -> set[str]:
|
||||
if not gated:
|
||||
return set()
|
||||
try:
|
||||
import yaml
|
||||
from hermes_constants import get_hermes_home
|
||||
config_path = str(get_hermes_home() / "config.yaml")
|
||||
if os.path.exists(config_path):
|
||||
with open(config_path, encoding="utf-8") as f:
|
||||
cfg = yaml.safe_load(f) or {}
|
||||
else:
|
||||
cfg = {}
|
||||
from hermes_cli.config import read_raw_config
|
||||
cfg = read_raw_config()
|
||||
except Exception:
|
||||
return set()
|
||||
result: set[str] = set()
|
||||
|
||||
@@ -146,15 +146,11 @@ def _get_command_timeout() -> int:
|
||||
``DEFAULT_COMMAND_TIMEOUT`` (30s) if unset or unreadable.
|
||||
"""
|
||||
try:
|
||||
hermes_home = get_hermes_home()
|
||||
config_path = hermes_home / "config.yaml"
|
||||
if config_path.exists():
|
||||
import yaml
|
||||
with open(config_path) as f:
|
||||
cfg = yaml.safe_load(f) or {}
|
||||
val = cfg.get("browser", {}).get("command_timeout")
|
||||
if val is not None:
|
||||
return max(int(val), 5) # Floor at 5s to avoid instant kills
|
||||
from hermes_cli.config import read_raw_config
|
||||
cfg = read_raw_config()
|
||||
val = cfg.get("browser", {}).get("command_timeout")
|
||||
if val is not None:
|
||||
return max(int(val), 5) # Floor at 5s to avoid instant kills
|
||||
except Exception as e:
|
||||
logger.debug("Could not read command_timeout from config: %s", e)
|
||||
return DEFAULT_COMMAND_TIMEOUT
|
||||
@@ -259,23 +255,19 @@ def _get_cloud_provider() -> Optional[CloudBrowserProvider]:
|
||||
|
||||
_cloud_provider_resolved = True
|
||||
try:
|
||||
hermes_home = get_hermes_home()
|
||||
config_path = hermes_home / "config.yaml"
|
||||
if config_path.exists():
|
||||
import yaml
|
||||
with open(config_path) as f:
|
||||
cfg = yaml.safe_load(f) or {}
|
||||
browser_cfg = cfg.get("browser", {})
|
||||
provider_key = None
|
||||
if isinstance(browser_cfg, dict) and "cloud_provider" in browser_cfg:
|
||||
provider_key = normalize_browser_cloud_provider(
|
||||
browser_cfg.get("cloud_provider")
|
||||
)
|
||||
if provider_key == "local":
|
||||
_cached_cloud_provider = None
|
||||
return None
|
||||
if provider_key and provider_key in _PROVIDER_REGISTRY:
|
||||
_cached_cloud_provider = _PROVIDER_REGISTRY[provider_key]()
|
||||
from hermes_cli.config import read_raw_config
|
||||
cfg = read_raw_config()
|
||||
browser_cfg = cfg.get("browser", {})
|
||||
provider_key = None
|
||||
if isinstance(browser_cfg, dict) and "cloud_provider" in browser_cfg:
|
||||
provider_key = normalize_browser_cloud_provider(
|
||||
browser_cfg.get("cloud_provider")
|
||||
)
|
||||
if provider_key == "local":
|
||||
_cached_cloud_provider = None
|
||||
return None
|
||||
if provider_key and provider_key in _PROVIDER_REGISTRY:
|
||||
_cached_cloud_provider = _PROVIDER_REGISTRY[provider_key]()
|
||||
except Exception as e:
|
||||
logger.debug("Could not read cloud_provider from config: %s", e)
|
||||
|
||||
@@ -326,13 +318,9 @@ def _allow_private_urls() -> bool:
|
||||
_allow_private_urls_resolved = True
|
||||
_cached_allow_private_urls = False # safe default
|
||||
try:
|
||||
hermes_home = get_hermes_home()
|
||||
config_path = hermes_home / "config.yaml"
|
||||
if config_path.exists():
|
||||
import yaml
|
||||
with open(config_path) as f:
|
||||
cfg = yaml.safe_load(f) or {}
|
||||
_cached_allow_private_urls = bool(cfg.get("browser", {}).get("allow_private_urls"))
|
||||
from hermes_cli.config import read_raw_config
|
||||
cfg = read_raw_config()
|
||||
_cached_allow_private_urls = bool(cfg.get("browser", {}).get("allow_private_urls"))
|
||||
except Exception as e:
|
||||
logger.debug("Could not read allow_private_urls from config: %s", e)
|
||||
return _cached_allow_private_urls
|
||||
@@ -1626,14 +1614,10 @@ def _maybe_start_recording(task_id: str):
|
||||
if task_id in _recording_sessions:
|
||||
return
|
||||
try:
|
||||
from hermes_cli.config import read_raw_config
|
||||
hermes_home = get_hermes_home()
|
||||
config_path = hermes_home / "config.yaml"
|
||||
record_enabled = False
|
||||
if config_path.exists():
|
||||
import yaml
|
||||
with open(config_path) as f:
|
||||
cfg = yaml.safe_load(f) or {}
|
||||
record_enabled = cfg.get("browser", {}).get("record_sessions", False)
|
||||
cfg = read_raw_config()
|
||||
record_enabled = cfg.get("browser", {}).get("record_sessions", False)
|
||||
|
||||
if not record_enabled:
|
||||
return
|
||||
|
||||
@@ -137,40 +137,36 @@ def _load_config_files() -> List[Dict[str, str]]:
|
||||
|
||||
result: List[Dict[str, str]] = []
|
||||
try:
|
||||
from hermes_cli.config import read_raw_config
|
||||
hermes_home = _resolve_hermes_home()
|
||||
config_path = hermes_home / "config.yaml"
|
||||
if config_path.exists():
|
||||
import yaml
|
||||
|
||||
with open(config_path) as f:
|
||||
cfg = yaml.safe_load(f) or {}
|
||||
cred_files = cfg.get("terminal", {}).get("credential_files")
|
||||
if isinstance(cred_files, list):
|
||||
hermes_home_resolved = hermes_home.resolve()
|
||||
for item in cred_files:
|
||||
if isinstance(item, str) and item.strip():
|
||||
rel = item.strip()
|
||||
if os.path.isabs(rel):
|
||||
logger.warning(
|
||||
"credential_files: rejected absolute config path %r", rel,
|
||||
)
|
||||
continue
|
||||
host_path = (hermes_home / rel).resolve()
|
||||
try:
|
||||
host_path.relative_to(hermes_home_resolved)
|
||||
except ValueError:
|
||||
logger.warning(
|
||||
"credential_files: rejected config path traversal %r "
|
||||
"(resolves to %s, outside HERMES_HOME %s)",
|
||||
rel, host_path, hermes_home_resolved,
|
||||
)
|
||||
continue
|
||||
if host_path.is_file():
|
||||
container_path = f"/root/.hermes/{rel}"
|
||||
result.append({
|
||||
"host_path": str(host_path),
|
||||
"container_path": container_path,
|
||||
})
|
||||
cfg = read_raw_config()
|
||||
cred_files = cfg.get("terminal", {}).get("credential_files")
|
||||
if isinstance(cred_files, list):
|
||||
hermes_home_resolved = hermes_home.resolve()
|
||||
for item in cred_files:
|
||||
if isinstance(item, str) and item.strip():
|
||||
rel = item.strip()
|
||||
if os.path.isabs(rel):
|
||||
logger.warning(
|
||||
"credential_files: rejected absolute config path %r", rel,
|
||||
)
|
||||
continue
|
||||
host_path = (hermes_home / rel).resolve()
|
||||
try:
|
||||
host_path.relative_to(hermes_home_resolved)
|
||||
except ValueError:
|
||||
logger.warning(
|
||||
"credential_files: rejected config path traversal %r "
|
||||
"(resolves to %s, outside HERMES_HOME %s)",
|
||||
rel, host_path, hermes_home_resolved,
|
||||
)
|
||||
continue
|
||||
if host_path.is_file():
|
||||
container_path = f"/root/.hermes/{rel}"
|
||||
result.append({
|
||||
"host_path": str(host_path),
|
||||
"container_path": container_path,
|
||||
})
|
||||
except Exception as e:
|
||||
logger.debug("Could not read terminal.credential_files from config: %s", e)
|
||||
|
||||
|
||||
@@ -66,18 +66,13 @@ def _load_config_passthrough() -> frozenset[str]:
|
||||
|
||||
result: set[str] = set()
|
||||
try:
|
||||
from hermes_constants import get_hermes_home
|
||||
config_path = get_hermes_home() / "config.yaml"
|
||||
if config_path.exists():
|
||||
import yaml
|
||||
|
||||
with open(config_path) as f:
|
||||
cfg = yaml.safe_load(f) or {}
|
||||
passthrough = cfg.get("terminal", {}).get("env_passthrough")
|
||||
if isinstance(passthrough, list):
|
||||
for item in passthrough:
|
||||
if isinstance(item, str) and item.strip():
|
||||
result.add(item.strip())
|
||||
from hermes_cli.config import read_raw_config
|
||||
cfg = read_raw_config()
|
||||
passthrough = cfg.get("terminal", {}).get("env_passthrough")
|
||||
if isinstance(passthrough, list):
|
||||
for item in passthrough:
|
||||
if isinstance(item, str) and item.strip():
|
||||
result.add(item.strip())
|
||||
except Exception as e:
|
||||
logger.debug("Could not read tools.env_passthrough from config: %s", e)
|
||||
|
||||
|
||||
@@ -98,12 +98,8 @@ def get_stt_model_from_config() -> Optional[str]:
|
||||
Silently returns ``None`` on any error (missing file, bad YAML, etc.).
|
||||
"""
|
||||
try:
|
||||
import yaml
|
||||
cfg_path = get_hermes_home() / "config.yaml"
|
||||
if cfg_path.exists():
|
||||
with open(cfg_path) as f:
|
||||
data = yaml.safe_load(f) or {}
|
||||
return data.get("stt", {}).get("model")
|
||||
from hermes_cli.config import read_raw_config
|
||||
return read_raw_config().get("stt", {}).get("model")
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
||||
Reference in New Issue
Block a user