fix(termux): harden execute_code and mobile browser/audio UX

This commit is contained in:
adybag14-cyber 2026-04-09 13:46:08 +02:00 committed by Teknium
parent 54d5138a54
commit 3237733ca5
10 changed files with 233 additions and 31 deletions

29
cli.py
View file

@ -1022,6 +1022,20 @@ def _is_termux_environment() -> bool:
return bool(os.getenv("TERMUX_VERSION") or "com.termux/files/usr" in prefix)
def _termux_example_image_path(filename: str = "cat.png") -> str:
"""Return a realistic example media path for the current Termux setup."""
candidates = [
os.path.expanduser("~/storage/shared"),
"/sdcard",
"/storage/emulated/0",
"/storage/self/primary",
]
for root in candidates:
if os.path.isdir(root):
return os.path.join(root, "Pictures", filename)
return os.path.join("~/storage/shared", "Pictures", filename)
def _split_path_input(raw: str) -> tuple[str, str]:
"""Split a leading file path token from trailing free-form text.
@ -3126,7 +3140,7 @@ class HermesCLI:
_cprint(
f" {_DIM}Clipboard image paste is not available on Termux — "
f"use /image <path> or paste a local image path like "
f"~/storage/shared/Pictures/cat.png{_RST}"
f"{_termux_example_image_path()}{_RST}"
)
return
@ -3144,7 +3158,7 @@ class HermesCLI:
"""Handle /image <path> — attach a local image file for the next prompt."""
raw_args = (cmd_original.split(None, 1)[1].strip() if " " in cmd_original else "")
if not raw_args:
hint = "~/storage/shared/Pictures/cat.png" if _is_termux_environment() else "/path/to/image.png"
hint = _termux_example_image_path() if _is_termux_environment() else "/path/to/image.png"
_cprint(f" {_DIM}Usage: /image <path> e.g. /image {hint}{_RST}")
return
@ -3162,7 +3176,7 @@ class HermesCLI:
if _remainder:
_cprint(f" {_DIM}Now type your prompt (or use --image in single-query mode): {_remainder}{_RST}")
elif _is_termux_environment():
_cprint(f" {_DIM}Tip: type your next message, or run hermes chat -q --image {image_path} \"What do you see?\"{_RST}")
_cprint(f" {_DIM}Tip: type your next message, or run hermes chat -q --image {_termux_example_image_path(image_path.name)} \"What do you see?\"{_RST}")
def _preprocess_images_with_vision(self, text: str, images: list, *, announce: bool = True) -> str:
"""Analyze attached images via the vision tool and return enriched text.
@ -3317,7 +3331,7 @@ class HermesCLI:
_cprint(f"\n {_DIM}Tip: Just type your message to chat with Hermes!{_RST}")
_cprint(f" {_DIM}Multi-line: Alt+Enter for a new line{_RST}")
if _is_termux_environment():
_cprint(f" {_DIM}Attach image: /image ~/storage/shared/Pictures/cat.png or start your prompt with a local image path{_RST}\n")
_cprint(f" {_DIM}Attach image: /image {_termux_example_image_path()} or start your prompt with a local image path{_RST}\n")
else:
_cprint(f" {_DIM}Paste image: Alt+V (or /paste){_RST}\n")
@ -6229,8 +6243,11 @@ class HermesCLI:
for line in reqs["details"].split("\n"):
_cprint(f" {_DIM}{line}{_RST}")
if reqs["missing_packages"]:
_cprint(f"\n {_BOLD}Install: pip install {' '.join(reqs['missing_packages'])}{_RST}")
_cprint(f" {_DIM}Or: pip install hermes-agent[voice]{_RST}")
if _is_termux_environment():
_cprint(f"\n {_BOLD}Install: pkg install python-numpy portaudio && python -m pip install sounddevice{_RST}")
else:
_cprint(f"\n {_BOLD}Install: pip install {' '.join(reqs['missing_packages'])}{_RST}")
_cprint(f" {_DIM}Or: pip install hermes-agent[voice]{_RST}")
return
with self._voice_lock:

View file

@ -596,7 +596,7 @@ def run_doctor(args):
else:
if _is_termux():
check_info("agent-browser is not installed (expected in the tested Termux path)")
check_info("Install it manually later with: npm install")
check_info("Install it manually later with: npm install -g agent-browser && agent-browser install")
else:
check_warn("agent-browser not installed", "(run: npm install)")
else:

View file

@ -5,6 +5,7 @@ from cli import (
HermesCLI,
_collect_query_images,
_format_image_attachment_badges,
_termux_example_image_path,
)
@ -80,6 +81,16 @@ class TestCollectQueryImages:
assert images == [img]
class TestTermuxImageHints:
def test_termux_example_image_path_prefers_real_shared_storage_root(self, monkeypatch):
existing = {"/sdcard", "/storage/emulated/0"}
monkeypatch.setattr("cli.os.path.isdir", lambda path: path in existing)
hint = _termux_example_image_path()
assert hint == "/sdcard/Pictures/cat.png"
class TestImageBadgeFormatting:
def test_compact_badges_use_filename_on_narrow_terminals(self, tmp_path):
img = _make_image(tmp_path / "Screenshot 2026-04-09 at 11.22.33 AM.png")

View file

@ -245,3 +245,46 @@ def test_run_doctor_termux_treats_docker_and_browser_warnings_as_expected(monkey
assert "Node.js not found (browser tools are optional in the tested Termux path)" in out
assert "Install Node.js on Termux with: pkg install nodejs" in out
assert "docker not found (optional)" not in out
def test_run_doctor_termux_does_not_mark_browser_available_without_agent_browser(monkeypatch, tmp_path):
home = tmp_path / ".hermes"
home.mkdir(parents=True, exist_ok=True)
(home / "config.yaml").write_text("memory: {}\n", encoding="utf-8")
project = tmp_path / "project"
project.mkdir(exist_ok=True)
monkeypatch.setenv("TERMUX_VERSION", "0.118.3")
monkeypatch.setenv("PREFIX", "/data/data/com.termux/files/usr")
monkeypatch.setattr(doctor_mod, "HERMES_HOME", home)
monkeypatch.setattr(doctor_mod, "PROJECT_ROOT", project)
monkeypatch.setattr(doctor_mod, "_DHH", str(home))
monkeypatch.setattr(doctor_mod.shutil, "which", lambda cmd: "/data/data/com.termux/files/usr/bin/node" if cmd in {"node", "npm"} else None)
fake_model_tools = types.SimpleNamespace(
check_tool_availability=lambda *a, **kw: (["terminal"], [{"name": "browser", "env_vars": [], "tools": ["browser_navigate"]}]),
TOOLSET_REQUIREMENTS={
"terminal": {"name": "terminal"},
"browser": {"name": "browser"},
},
)
monkeypatch.setitem(sys.modules, "model_tools", fake_model_tools)
try:
from hermes_cli import auth as _auth_mod
monkeypatch.setattr(_auth_mod, "get_nous_auth_status", lambda: {})
monkeypatch.setattr(_auth_mod, "get_codex_auth_status", lambda: {})
except Exception:
pass
import io, contextlib
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
doctor_mod.run_doctor(Namespace(fix=False))
out = buf.getvalue()
assert "✓ browser" not in out
assert "browser" in out
assert "system dependency not met" in out
assert "agent-browser is not installed (expected in the tested Termux path)" in out
assert "npm install -g agent-browser && agent-browser install" in out

View file

@ -13,6 +13,7 @@ from tools.browser_tool import (
_find_agent_browser,
_run_browser_command,
_SANE_PATH,
check_browser_requirements,
)
@ -149,6 +150,17 @@ class TestFindAgentBrowser:
_find_agent_browser()
class TestBrowserRequirements:
def test_termux_requires_real_agent_browser_install_not_npx_fallback(self, monkeypatch):
monkeypatch.setenv("TERMUX_VERSION", "0.118.3")
monkeypatch.setenv("PREFIX", "/data/data/com.termux/files/usr")
monkeypatch.setattr("tools.browser_tool._is_camofox_mode", lambda: False)
monkeypatch.setattr("tools.browser_tool._get_cloud_provider", lambda: None)
monkeypatch.setattr("tools.browser_tool._find_agent_browser", lambda: "npx agent-browser")
assert check_browser_requirements() is False
class TestRunBrowserCommandPathConstruction:
"""Verify _run_browser_command() includes Homebrew node dirs in subprocess PATH."""

View file

@ -44,6 +44,7 @@ from tools.code_execution_tool import (
build_execute_code_schema,
EXECUTE_CODE_SCHEMA,
_TOOL_DOC_LINES,
_execute_remote,
)
@ -115,6 +116,48 @@ class TestHermesToolsGeneration(unittest.TestCase):
self.assertIn("def retry(", src)
self.assertIn("import json, os, socket, shlex, time", src)
def test_file_transport_uses_tempfile_fallback_for_rpc_dir(self):
src = generate_hermes_tools_module(["terminal"], transport="file")
self.assertIn("import json, os, shlex, tempfile, time", src)
self.assertIn("os.path.join(tempfile.gettempdir(), \"hermes_rpc\")", src)
self.assertNotIn('os.environ.get("HERMES_RPC_DIR", "/tmp/hermes_rpc")', src)
class TestExecuteCodeRemoteTempDir(unittest.TestCase):
def test_execute_remote_uses_backend_temp_dir_for_sandbox(self):
class FakeEnv:
def __init__(self):
self.commands = []
def get_temp_dir(self):
return "/data/data/com.termux/files/usr/tmp"
def execute(self, command, cwd=None, timeout=None):
self.commands.append((command, cwd, timeout))
if "command -v python3" in command:
return {"output": "OK\n"}
if "python3 script.py" in command:
return {"output": "hello\n", "returncode": 0}
return {"output": ""}
env = FakeEnv()
fake_thread = MagicMock()
with patch("tools.code_execution_tool._load_config", return_value={"timeout": 30, "max_tool_calls": 5}), \
patch("tools.code_execution_tool._get_or_create_env", return_value=(env, "ssh")), \
patch("tools.code_execution_tool._ship_file_to_remote"), \
patch("tools.code_execution_tool.threading.Thread", return_value=fake_thread):
result = json.loads(_execute_remote("print('hello')", "task-1", ["terminal"]))
self.assertEqual(result["status"], "success")
mkdir_cmd = env.commands[1][0]
run_cmd = next(cmd for cmd, _, _ in env.commands if "python3 script.py" in cmd)
cleanup_cmd = env.commands[-1][0]
self.assertIn("mkdir -p /data/data/com.termux/files/usr/tmp/hermes_exec_", mkdir_cmd)
self.assertIn("HERMES_RPC_DIR=/data/data/com.termux/files/usr/tmp/hermes_exec_", run_cmd)
self.assertIn("rm -rf /data/data/com.termux/files/usr/tmp/hermes_exec_", cleanup_cmd)
self.assertNotIn("mkdir -p /tmp/hermes_exec_", mkdir_cmd)
@unittest.skipIf(sys.platform == "win32", "UDS not available on Windows")
class TestExecuteCode(unittest.TestCase):

View file

@ -183,6 +183,21 @@ class TestDetectAudioEnvironment:
assert result["available"] is False
assert any("PortAudio" in w for w in result["warnings"])
def test_termux_import_error_shows_termux_install_guidance(self, monkeypatch):
monkeypatch.setenv("TERMUX_VERSION", "0.118.3")
monkeypatch.setenv("PREFIX", "/data/data/com.termux/files/usr")
monkeypatch.delenv("SSH_CLIENT", raising=False)
monkeypatch.delenv("SSH_TTY", raising=False)
monkeypatch.delenv("SSH_CONNECTION", raising=False)
monkeypatch.setattr("tools.voice_mode._import_audio", lambda: (_ for _ in ()).throw(ImportError("no audio libs")))
from tools.voice_mode import detect_audio_environment
result = detect_audio_environment()
assert result["available"] is False
assert any("pkg install python-numpy portaudio" in w for w in result["warnings"])
assert any("python -m pip install sounddevice" in w for w in result["warnings"])
# ============================================================================
# check_voice_requirements

View file

@ -285,6 +285,17 @@ def _get_cloud_provider() -> Optional[CloudBrowserProvider]:
return _cached_cloud_provider
def _is_termux_environment() -> bool:
prefix = os.getenv("PREFIX", "")
return bool(os.getenv("TERMUX_VERSION") or "com.termux/files/usr" in prefix)
def _browser_install_hint() -> str:
if _is_termux_environment():
return "npm install -g agent-browser && agent-browser install"
return "npm install -g agent-browser && agent-browser install --with-deps"
def _is_local_mode() -> bool:
"""Return True when the browser tool will use a local browser backend."""
if _get_cdp_override():
@ -796,7 +807,8 @@ def _find_agent_browser() -> str:
return "npx agent-browser"
raise FileNotFoundError(
"agent-browser CLI not found. Install it with: npm install -g agent-browser\n"
"agent-browser CLI not found. Install it with: "
f"{_browser_install_hint()}\n"
"Or run 'npm install' in the repo root to install locally.\n"
"Or ensure npx is available in your PATH."
)
@ -2040,10 +2052,17 @@ def check_browser_requirements() -> bool:
# The agent-browser CLI is always required
try:
_find_agent_browser()
browser_cmd = _find_agent_browser()
except FileNotFoundError:
return False
# On Termux, the bare npx fallback is too fragile to treat as a satisfied
# local browser dependency. Require a real install (global or local) so the
# browser tool is not advertised as available when it will likely fail on
# first use.
if _is_termux_environment() and _is_local_mode() and browser_cmd.strip() == "npx agent-browser":
return False
# In cloud mode, also require provider credentials
provider = _get_cloud_provider()
if provider is not None and not provider.is_configured():

View file

@ -33,6 +33,7 @@ import json
import logging
import os
import platform
import shlex
import signal
import socket
import subprocess
@ -246,9 +247,9 @@ def _call(tool_name, args):
_FILE_TRANSPORT_HEADER = '''\
"""Auto-generated Hermes tools RPC stubs (file-based transport)."""
import json, os, shlex, time
import json, os, shlex, tempfile, time
_RPC_DIR = os.environ.get("HERMES_RPC_DIR", "/tmp/hermes_rpc")
_RPC_DIR = os.environ.get("HERMES_RPC_DIR") or os.path.join(tempfile.gettempdir(), "hermes_rpc")
_seq = 0
''' + _COMMON_HELPERS + '''\
@ -536,13 +537,30 @@ def _ship_file_to_remote(env, remote_path: str, content: str) -> None:
quotes are fine.
"""
encoded = base64.b64encode(content.encode("utf-8")).decode("ascii")
quoted_remote_path = shlex.quote(remote_path)
env.execute(
f"echo '{encoded}' | base64 -d > {remote_path}",
f"echo '{encoded}' | base64 -d > {quoted_remote_path}",
cwd="/",
timeout=30,
)
def _env_temp_dir(env: Any) -> str:
"""Return a writable temp dir for env-backed execute_code sandboxes."""
get_temp_dir = getattr(env, "get_temp_dir", None)
if callable(get_temp_dir):
try:
temp_dir = get_temp_dir()
if isinstance(temp_dir, str) and temp_dir.startswith("/"):
return temp_dir.rstrip("/") or "/"
except Exception as exc:
logger.debug("Could not resolve execute_code env temp dir: %s", exc)
candidate = tempfile.gettempdir()
if isinstance(candidate, str) and candidate.startswith("/"):
return candidate.rstrip("/") or "/"
return "/tmp"
def _rpc_poll_loop(
env,
rpc_dir: str,
@ -563,11 +581,12 @@ def _rpc_poll_loop(
poll_interval = 0.1 # 100 ms
quoted_rpc_dir = shlex.quote(rpc_dir)
while not stop_event.is_set():
try:
# List pending request files (skip .tmp partials)
ls_result = env.execute(
f"ls -1 {rpc_dir}/req_* 2>/dev/null || true",
f"ls -1 {quoted_rpc_dir}/req_* 2>/dev/null || true",
cwd="/",
timeout=10,
)
@ -589,9 +608,10 @@ def _rpc_poll_loop(
call_start = time.monotonic()
quoted_req_file = shlex.quote(req_file)
# Read request
read_result = env.execute(
f"cat {req_file}",
f"cat {quoted_req_file}",
cwd="/",
timeout=10,
)
@ -600,7 +620,7 @@ def _rpc_poll_loop(
except (json.JSONDecodeError, ValueError):
logger.debug("Malformed RPC request in %s", req_file)
# Remove bad request to avoid infinite retry
env.execute(f"rm -f {req_file}", cwd="/", timeout=5)
env.execute(f"rm -f {quoted_req_file}", cwd="/", timeout=5)
continue
tool_name = request.get("tool", "")
@ -608,6 +628,7 @@ def _rpc_poll_loop(
seq = request.get("seq", 0)
seq_str = f"{seq:06d}"
res_file = f"{rpc_dir}/res_{seq_str}"
quoted_res_file = shlex.quote(res_file)
# Enforce allow-list
if tool_name not in allowed_tools:
@ -665,14 +686,14 @@ def _rpc_poll_loop(
tool_result.encode("utf-8")
).decode("ascii")
env.execute(
f"echo '{encoded_result}' | base64 -d > {res_file}.tmp"
f" && mv {res_file}.tmp {res_file}",
f"echo '{encoded_result}' | base64 -d > {quoted_res_file}.tmp"
f" && mv {quoted_res_file}.tmp {quoted_res_file}",
cwd="/",
timeout=60,
)
# Remove the request file
env.execute(f"rm -f {req_file}", cwd="/", timeout=5)
env.execute(f"rm -f {quoted_req_file}", cwd="/", timeout=5)
except Exception as e:
if not stop_event.is_set():
@ -707,7 +728,10 @@ def _execute_remote(
env, env_type = _get_or_create_env(effective_task_id)
sandbox_id = uuid.uuid4().hex[:12]
sandbox_dir = f"/tmp/hermes_exec_{sandbox_id}"
temp_dir = _env_temp_dir(env)
sandbox_dir = f"{temp_dir}/hermes_exec_{sandbox_id}"
quoted_sandbox_dir = shlex.quote(sandbox_dir)
quoted_rpc_dir = shlex.quote(f"{sandbox_dir}/rpc")
tool_call_log: list = []
tool_call_counter = [0]
@ -735,7 +759,7 @@ def _execute_remote(
# Create sandbox directory on remote
env.execute(
f"mkdir -p {sandbox_dir}/rpc", cwd="/", timeout=10,
f"mkdir -p {quoted_rpc_dir}", cwd="/", timeout=10,
)
# Generate and ship files
@ -759,7 +783,7 @@ def _execute_remote(
# Build environment variable prefix for the script
env_prefix = (
f"HERMES_RPC_DIR={sandbox_dir}/rpc "
f"HERMES_RPC_DIR={shlex.quote(f'{sandbox_dir}/rpc')} "
f"PYTHONDONTWRITEBYTECODE=1"
)
tz = os.getenv("HERMES_TIMEZONE", "").strip()
@ -770,7 +794,7 @@ def _execute_remote(
logger.info("Executing code on %s backend (task %s)...",
env_type, effective_task_id[:8])
script_result = env.execute(
f"cd {sandbox_dir} && {env_prefix} python3 script.py",
f"cd {quoted_sandbox_dir} && {env_prefix} python3 script.py",
timeout=timeout,
)
@ -807,7 +831,7 @@ def _execute_remote(
# Clean up remote sandbox dir
try:
env.execute(
f"rm -rf {sandbox_dir}", cwd="/", timeout=15,
f"rm -rf {quoted_sandbox_dir}", cwd="/", timeout=15,
)
except Exception:
logger.debug("Failed to clean up remote sandbox %s", sandbox_dir)

View file

@ -48,6 +48,17 @@ def _audio_available() -> bool:
return False
def _is_termux_environment() -> bool:
prefix = os.getenv("PREFIX", "")
return bool(os.getenv("TERMUX_VERSION") or "com.termux/files/usr" in prefix)
def _voice_capture_install_hint() -> str:
if _is_termux_environment():
return "pkg install python-numpy portaudio && python -m pip install sounddevice"
return "pip install sounddevice numpy"
def detect_audio_environment() -> dict:
"""Detect if the current environment supports audio I/O.
@ -98,14 +109,21 @@ def detect_audio_environment() -> dict:
else:
warnings.append("Audio subsystem error (PortAudio cannot query devices)")
except ImportError:
warnings.append("Audio libraries not installed (pip install sounddevice numpy)")
warnings.append(f"Audio libraries not installed ({_voice_capture_install_hint()})")
except OSError:
warnings.append(
"PortAudio system library not found -- install it first:\n"
" Linux: sudo apt-get install libportaudio2\n"
" macOS: brew install portaudio\n"
"Then retry /voice on."
)
if _is_termux_environment():
warnings.append(
"PortAudio system library not found -- install it first:\n"
" Termux: pkg install portaudio\n"
"Then retry /voice on."
)
else:
warnings.append(
"PortAudio system library not found -- install it first:\n"
" Linux: sudo apt-get install libportaudio2\n"
" macOS: brew install portaudio\n"
"Then retry /voice on."
)
return {
"available": not warnings,
@ -748,7 +766,7 @@ def check_voice_requirements() -> Dict[str, Any]:
if has_audio:
details_parts.append("Audio capture: OK")
else:
details_parts.append("Audio capture: MISSING (pip install sounddevice numpy)")
details_parts.append(f"Audio capture: MISSING ({_voice_capture_install_hint()})")
if not stt_enabled:
details_parts.append("STT provider: DISABLED in config (stt.enabled: false)")