Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -216,7 +216,11 @@ def realign_markdown_tables(*args, **kwargs):

# Load .env from ~/.hermes/.env first, then project root as dev fallback.
# User-managed env files should override stale shell exports on restart.
from hermes_constants import get_hermes_home, display_hermes_home
from hermes_constants import (
apply_configured_ipv4_preference,
display_hermes_home,
get_hermes_home,
)
from hermes_cli.browser_connect import (
DEFAULT_BROWSER_CDP_URL,
is_browser_debug_ready,
Expand All @@ -227,6 +231,8 @@ def realign_markdown_tables(*args, **kwargs):
from utils import base_url_host_matches, fast_safe_load

_hermes_home = get_hermes_home()
# Honor network.force_ipv4 for the legacy `python cli.py` entry path.
apply_configured_ipv4_preference(hermes_home=_hermes_home)
_project_env = Path(__file__).parent / '.env'
load_hermes_dotenv(hermes_home=_hermes_home, project_env=_project_env)

Expand Down
47 changes: 47 additions & 0 deletions hermes_constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -1355,6 +1355,53 @@ def _ipv4_getaddrinfo(host, port, family=0, type=0, proto=0, flags=0):
socket.getaddrinfo = _ipv4_getaddrinfo # type: ignore[assignment]


def apply_configured_ipv4_preference(hermes_home: Path | None = None) -> bool:
"""Apply ``network.force_ipv4`` from ``config.yaml`` when enabled.

Used by entry points that do not go through ``hermes_cli.main`` bootstrap
(``run_agent``, legacy ``cli``). Best-effort and never raises.

Honors managed-scope overlays so an administrator-pinned
``network.force_ipv4`` wins the same way the canonical CLI bootstrap does.
"""
try:
import yaml
except Exception:
return False

config_path = (hermes_home or get_hermes_home()) / "config.yaml"
try:
with config_path.open(encoding="utf-8") as fh:
config = yaml.safe_load(fh)
except OSError:
return False
except Exception:
return False

# yaml.safe_load may return a scalar/list for valid YAML — guard before .get
if not isinstance(config, dict):
return False

# Match hermes_cli/main.py: managed overlay before reading force_ipv4.
# Fail-open so a managed-scope import failure never blocks startup.
try:
from hermes_cli import managed_scope

config = managed_scope.apply_managed_overlay(config)
except Exception:
pass

if not isinstance(config, dict):
return False

network_cfg = config.get("network", {})
if not isinstance(network_cfg, dict):
return False
force = bool(network_cfg.get("force_ipv4"))
apply_ipv4_preference(force=force)
return force


# ─── Streaming Response Constants ────────────────────────────────────────────

# Response ID for partial stream stubs used during error recovery
Expand Down
5 changes: 4 additions & 1 deletion run_agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,7 @@
from pathlib import Path
from types import SimpleNamespace

from hermes_constants import get_hermes_home
from hermes_constants import get_hermes_home, apply_configured_ipv4_preference


def _launch_cwd_for_session(source: str) -> Optional[str]:
Expand Down Expand Up @@ -125,6 +125,9 @@ def _session_source_for_agent(platform: Optional[str]) -> str:
)

_hermes_home = get_hermes_home()
# Honor network.force_ipv4 for the hermes-agent / library entry path
# (hermes_cli.main already does this for the `hermes` console script).
apply_configured_ipv4_preference(hermes_home=_hermes_home)
_project_env = Path(__file__).parent / '.env'
_loaded_env_paths = load_hermes_dotenv(hermes_home=_hermes_home, project_env=_project_env)
if _loaded_env_paths:
Expand Down
157 changes: 153 additions & 4 deletions tests/test_ipv4_preference.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

import importlib
import socket

import sys


def _reload_constants():
Expand All @@ -23,7 +23,6 @@ def teardown_method(self):
"""Restore the original getaddrinfo after each test."""
socket.getaddrinfo = self._original


def test_patches_getaddrinfo_when_forced(self):
"""Patches socket.getaddrinfo when force=True."""
from hermes_constants import apply_ipv4_preference
Expand All @@ -45,7 +44,6 @@ def test_af_unspec_becomes_af_inet(self):
from hermes_constants import apply_ipv4_preference

calls = []
original = socket.getaddrinfo

def mock_getaddrinfo(host, port, family=0, type=0, proto=0, flags=0):
calls.append(family)
Expand All @@ -63,7 +61,6 @@ def test_explicit_family_preserved(self):
from hermes_constants import apply_ipv4_preference

calls = []
original = socket.getaddrinfo

def mock_getaddrinfo(host, port, family=0, type=0, proto=0, flags=0):
calls.append(family)
Expand All @@ -76,4 +73,156 @@ def mock_getaddrinfo(host, port, family=0, type=0, proto=0, flags=0):
assert calls[-1] == socket.AF_INET6, "Explicit AF_INET6 should pass through"


class TestApplyConfiguredIPv4Preference:
"""Tests for apply_configured_ipv4_preference()."""

def test_returns_false_when_config_missing(self, monkeypatch, tmp_path):
hermes_constants = _reload_constants()
calls = []
monkeypatch.setattr(
hermes_constants,
"apply_ipv4_preference",
lambda force=False: calls.append(force),
)

assert hermes_constants.apply_configured_ipv4_preference(tmp_path) is False
assert calls == []

def test_reads_force_ipv4_from_config(self, monkeypatch, tmp_path):
hermes_constants = _reload_constants()
(tmp_path / "config.yaml").write_text(
"network:\n force_ipv4: true\n", encoding="utf-8"
)
calls = []
monkeypatch.setattr(
hermes_constants,
"apply_ipv4_preference",
lambda force=False: calls.append(force),
)

assert hermes_constants.apply_configured_ipv4_preference(tmp_path) is True
assert calls == [True]

def test_ignores_invalid_network_section(self, monkeypatch, tmp_path):
hermes_constants = _reload_constants()
(tmp_path / "config.yaml").write_text("network: enabled\n", encoding="utf-8")
calls = []
monkeypatch.setattr(
hermes_constants,
"apply_ipv4_preference",
lambda force=False: calls.append(force),
)

assert hermes_constants.apply_configured_ipv4_preference(tmp_path) is False
assert calls == []

def test_scalar_root_config_is_safe(self, monkeypatch, tmp_path):
"""Non-mapping YAML roots must not raise during import-time bootstrap."""
hermes_constants = _reload_constants()
(tmp_path / "config.yaml").write_text("true\n", encoding="utf-8")
calls = []
monkeypatch.setattr(
hermes_constants,
"apply_ipv4_preference",
lambda force=False: calls.append(force),
)

assert hermes_constants.apply_configured_ipv4_preference(tmp_path) is False
assert calls == []

def test_list_root_config_is_safe(self, monkeypatch, tmp_path):
hermes_constants = _reload_constants()
(tmp_path / "config.yaml").write_text("- a\n- b\n", encoding="utf-8")
calls = []
monkeypatch.setattr(
hermes_constants,
"apply_ipv4_preference",
lambda force=False: calls.append(force),
)

assert hermes_constants.apply_configured_ipv4_preference(tmp_path) is False
assert calls == []

def test_managed_scope_overlay_wins(self, monkeypatch, tmp_path):
"""Administrator-pinned network.force_ipv4 must override user config."""
hermes_constants = _reload_constants()
user_home = tmp_path / "user"
managed_dir = tmp_path / "managed"
user_home.mkdir()
managed_dir.mkdir()
(user_home / "config.yaml").write_text(
"network:\n force_ipv4: false\n", encoding="utf-8"
)
(managed_dir / "config.yaml").write_text(
"network:\n force_ipv4: true\n", encoding="utf-8"
)

monkeypatch.setenv("HERMES_MANAGED_DIR", str(managed_dir))
# Clear managed-scope caches so the override is picked up.
from hermes_cli import managed_scope
managed_scope.invalidate_managed_cache()

calls = []
monkeypatch.setattr(
hermes_constants,
"apply_ipv4_preference",
lambda force=False: calls.append(force),
)

assert hermes_constants.apply_configured_ipv4_preference(user_home) is True
assert calls == [True]


class TestBootstrapWiring:
"""Entry points should apply the config-driven IPv4 preference on import."""

def setup_method(self):
self._saved = {name: sys.modules.get(name) for name in ("cli", "run_agent")}

def teardown_method(self):
for name, module in self._saved.items():
if module is None:
sys.modules.pop(name, None)
else:
sys.modules[name] = module

def test_run_agent_bootstrap_applies_configured_ipv4(self, monkeypatch):
import hermes_constants

calls = []
monkeypatch.setattr(
hermes_constants,
"apply_configured_ipv4_preference",
lambda hermes_home=None: calls.append(hermes_home),
)
sys.modules.pop("run_agent", None)

run_agent = importlib.import_module("run_agent")

assert calls == [run_agent._hermes_home]

def test_cli_bootstrap_applies_configured_ipv4(self, monkeypatch):
import hermes_constants

calls = []
monkeypatch.setattr(
hermes_constants,
"apply_configured_ipv4_preference",
lambda hermes_home=None: calls.append(hermes_home),
)
sys.modules.pop("cli", None)

cli = importlib.import_module("cli")

assert calls == [cli._hermes_home]


class TestConfigDefault:
"""Verify network section exists in DEFAULT_CONFIG."""

def test_network_force_ipv4_default_is_false(self):
from hermes_cli.config_defaults import DEFAULT_CONFIG

network = DEFAULT_CONFIG.get("network", {})
assert isinstance(network, dict)
assert network.get("force_ipv4") is False