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

refactor(cli): container exec routing moves out of main.py into main_container.py

Browse Source 
main.py may only shrink vs main (code-health FILE_LINES); this branch grew it
by 9 lines. _probe_container and _exec_in_container (the NixOS container-mode
re-exec, 96 lines) move verbatim to the topical sibling
hermes_cli/main_container.py; the entry point is now public
(exec_in_container) and main() late-imports it at its only call site. The
test import and the managed-runtime resolution allowlist entry follow the
move. No re-export shim. Behaviour is unchanged.
This commit is contained in:
Teknium
2026-10-05 00:18:44 -07:00
parent b82d35571c
commit 00ba600fc0
4 changed files with 112 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

100
hermes_cli/main.py
View File

@@ -1352,102 +1352,6 @@ def _resolve_last_session(source: str = "cli") -> Optional[str]:
return None
def _probe_container(cmd: list, backend: str, via_sudo: bool = False):
"""Run a container inspect probe, returning the CompletedProcess.
Catches TimeoutExpired specifically for a human-readable message;
all other exceptions propagate naturally.
"""
try:
return subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=15)
except subprocess.TimeoutExpired:
label = f"sudo {backend}" if via_sudo else backend
print(
f"Error: timed out waiting for {label} to respond.\n"
f"The {backend} daemon may be unresponsive or starting up.",
file=sys.stderr,
)
sys.exit(1)
def _exec_in_container(container_info: dict, cli_args: list):
"""Replace the current process with a command inside the managed container.
Probes whether sudo is needed (rootful containers), then os.execvp
into the container. On success the Python process is replaced entirely
and the container's exit code becomes the process exit code (OS semantics).
On failure, OSError propagates naturally.
Args:
container_info: dict with backend, container_name, exec_user, hermes_bin
cli_args: the original CLI arguments (everything after 'hermes')
"""
backend = container_info["backend"]
container_name = container_info["container_name"]
exec_user = container_info["exec_user"]
hermes_bin = container_info["hermes_bin"]
runtime = shutil.which(backend)
if not runtime:
print(
f"Error: {backend} not found on PATH. Cannot route to container.",
file=sys.stderr,
)
sys.exit(1)
# Rootful containers (NixOS systemd service) are invisible to unprivileged
# users — Podman uses per-user namespaces, Docker needs group access.
# Probe whether the runtime can see the container; if not, try via sudo.
inspect_cmd = [runtime, "inspect", "--format", "ok", container_name]
cmd_prefix = [runtime]
if _probe_container(inspect_cmd, backend).returncode != 0:
sudo_path = shutil.which("sudo")
if not sudo_path:
print(
f"Error: container '{container_name}' not found via {backend}.\n"
f"The container may be running under root. Try: sudo hermes {' '.join(cli_args)}",
file=sys.stderr,
)
sys.exit(1)
cmd_prefix = [sudo_path, "-n", runtime]
if _probe_container(cmd_prefix[:2] + inspect_cmd, backend, via_sudo=True).returncode != 0:
print(
f"Error: container '{container_name}' not found via {backend}.\n"
f"\n"
f"The container is likely running as root. Your user cannot see it\n"
f"because {backend} uses per-user namespaces. Grant passwordless\n"
f"sudo for {backend} — the -n (non-interactive) flag is required\n"
f"because a password prompt would hang or break piped commands.\n"
f"\n"
f"On NixOS:\n"
f"\n"
f" security.sudo.extraRules = [{{\n"
f' users = [ "{os.getenv("USER", "your-user")}" ];\n'
f' commands = [{{ command = "{runtime}"; options = [ "NOPASSWD" ]; }}];\n'
f" }}];\n"
f"\n"
f"Or run: sudo hermes {' '.join(cli_args)}",
file=sys.stderr,
)
sys.exit(1)
env_flags = []
for var in ("TERM", "COLORTERM", "LANG", "LC_ALL"):
val = os.environ.get(var)
if val:
env_flags.extend(["-e", f"{var}={val}"])
exec_cmd = (
cmd_prefix
+ ["exec", "-it" if sys.stdin.isatty() else "-i", "-u", exec_user]
+ env_flags
+ [container_name, hermes_bin]
+ cli_args
)
os.execvp(exec_cmd[0], exec_cmd)
def _resolve_session_by_name_or_id(name_or_id: str) -> Optional[str]:
"""Resolve a session title or ID to a session ID (None if neither matches).
@@ -3695,7 +3599,9 @@ def main():
container_info = get_container_exec_info()
if container_info:
_exec_in_container(container_info, sys.argv[1:])
from hermes_cli.main_container import exec_in_container
exec_in_container(container_info, sys.argv[1:])
sys.exit(1) # unreachable: execvp replaces the process or raises
args = _parse_cli_args(parser, subparsers, sys.argv[1:])

104
hermes_cli/main_container.py Normal file
View File

@@ -0,0 +1,104 @@
"""NixOS container mode: route a ``hermes`` invocation into the managed container (``main()``)."""
from __future__ import annotations
import os
import shutil
import subprocess
import sys
def _probe_container(cmd: list, backend: str, via_sudo: bool = False):
"""Run a container inspect probe, returning the CompletedProcess.
Catches TimeoutExpired specifically for a human-readable message;
all other exceptions propagate naturally.
"""
try:
return subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=15)
except subprocess.TimeoutExpired:
label = f"sudo {backend}" if via_sudo else backend
print(
f"Error: timed out waiting for {label} to respond.\n"
f"The {backend} daemon may be unresponsive or starting up.",
file=sys.stderr,
)
sys.exit(1)
def exec_in_container(container_info: dict, cli_args: list):
"""Replace the current process with a command inside the managed container.
Probes whether sudo is needed (rootful containers), then os.execvp
into the container. On success the Python process is replaced entirely
and the container's exit code becomes the process exit code (OS semantics).
On failure, OSError propagates naturally.
Args:
container_info: dict with backend, container_name, exec_user, hermes_bin
cli_args: the original CLI arguments (everything after 'hermes')
"""
backend = container_info["backend"]
container_name = container_info["container_name"]
exec_user = container_info["exec_user"]
hermes_bin = container_info["hermes_bin"]
runtime = shutil.which(backend)
if not runtime:
print(
f"Error: {backend} not found on PATH. Cannot route to container.",
file=sys.stderr,
)
sys.exit(1)
# Rootful containers (NixOS systemd service) are invisible to unprivileged
# users — Podman uses per-user namespaces, Docker needs group access.
# Probe whether the runtime can see the container; if not, try via sudo.
inspect_cmd = [runtime, "inspect", "--format", "ok", container_name]
cmd_prefix = [runtime]
if _probe_container(inspect_cmd, backend).returncode != 0:
sudo_path = shutil.which("sudo")
if not sudo_path:
print(
f"Error: container '{container_name}' not found via {backend}.\n"
f"The container may be running under root. Try: sudo hermes {' '.join(cli_args)}",
file=sys.stderr,
)
sys.exit(1)
cmd_prefix = [sudo_path, "-n", runtime]
if _probe_container(cmd_prefix[:2] + inspect_cmd, backend, via_sudo=True).returncode != 0:
print(
f"Error: container '{container_name}' not found via {backend}.\n"
f"\n"
f"The container is likely running as root. Your user cannot see it\n"
f"because {backend} uses per-user namespaces. Grant passwordless\n"
f"sudo for {backend} — the -n (non-interactive) flag is required\n"
f"because a password prompt would hang or break piped commands.\n"
f"\n"
f"On NixOS:\n"
f"\n"
f" security.sudo.extraRules = [{{\n"
f' users = [ "{os.getenv("USER", "your-user")}" ];\n'
f' commands = [{{ command = "{runtime}"; options = [ "NOPASSWD" ]; }}];\n'
f" }}];\n"
f"\n"
f"Or run: sudo hermes {' '.join(cli_args)}",
file=sys.stderr,
)
sys.exit(1)
env_flags = []
for var in ("TERM", "COLORTERM", "LANG", "LC_ALL"):
val = os.environ.get(var)
if val:
env_flags.extend(["-e", f"{var}={val}"])
exec_cmd = (
cmd_prefix
+ ["exec", "-it" if sys.stdin.isatty() else "-i", "-u", exec_user]
+ env_flags
+ [container_name, hermes_bin]
+ cli_args
)
os.execvp(exec_cmd[0], exec_cmd)

4
tests/fixtures/resolution_allowlist.json vendored
View File

@@ -250,8 +250,8 @@
"kind": "bare_which"
},
{
"path": "hermes_cli/main.py",
"symbol": "_exec_in_container",
"path": "hermes_cli/main_container.py",
"symbol": "exec_in_container",
"kind": "bare_which"
},
{

6
tests/hermes_cli/test_container_aware_cli.py
View File

@@ -51,7 +51,7 @@ def test_get_container_exec_info_returns_metadata(container_env):
# =============================================================================
# _exec_in_container
# exec_in_container (hermes_cli/main_container.py)
# =============================================================================
@@ -68,7 +68,7 @@ def docker_container_info():
def test_exec_in_container_calls_execvp(docker_container_info):
"""Verifies os.execvp is called with correct args: runtime, tty flags,
user, env vars, container name, binary, and CLI args."""
from hermes_cli.main import _exec_in_container
from hermes_cli.main_container import exec_in_container
with patch("shutil.which", return_value="/usr/bin/docker"), \
patch("subprocess.run") as mock_run, \
@@ -79,7 +79,7 @@ def test_exec_in_container_calls_execvp(docker_container_info):
mock_stdin.isatty.return_value = True
mock_run.return_value = MagicMock(returncode=0)
_exec_in_container(docker_container_info, ["chat", "-m", "opus"])
exec_in_container(docker_container_info, ["chat", "-m", "opus"])
mock_execvp.assert_called_once()
cmd = mock_execvp.call_args[0][1]