fix(termux): honor temp dirs for local temp artifacts

This commit is contained in:
adybag14-cyber 2026-04-09 08:22:55 +02:00 committed by Teknium
parent e79cc88985
commit 122925a6f2
5 changed files with 151 additions and 7 deletions

View file

@ -0,0 +1,51 @@
from unittest.mock import patch
from tools.environments.local import LocalEnvironment
class TestLocalTempDir:
def test_uses_os_tmpdir_for_session_artifacts(self, monkeypatch):
monkeypatch.setenv("TMPDIR", "/data/data/com.termux/files/usr/tmp")
monkeypatch.delenv("TMP", raising=False)
monkeypatch.delenv("TEMP", raising=False)
with patch.object(LocalEnvironment, "init_session", autospec=True, return_value=None):
env = LocalEnvironment(cwd=".", timeout=10)
assert env.get_temp_dir() == "/data/data/com.termux/files/usr/tmp"
assert env._snapshot_path == f"/data/data/com.termux/files/usr/tmp/hermes-snap-{env._session_id}.sh"
assert env._cwd_file == f"/data/data/com.termux/files/usr/tmp/hermes-cwd-{env._session_id}.txt"
def test_prefers_backend_env_tmpdir_override(self, monkeypatch):
monkeypatch.delenv("TMPDIR", raising=False)
monkeypatch.delenv("TMP", raising=False)
monkeypatch.delenv("TEMP", raising=False)
with patch.object(LocalEnvironment, "init_session", autospec=True, return_value=None):
env = LocalEnvironment(
cwd=".",
timeout=10,
env={"TMPDIR": "/data/data/com.termux/files/home/.cache/hermes-tmp/"},
)
assert env.get_temp_dir() == "/data/data/com.termux/files/home/.cache/hermes-tmp"
assert env._snapshot_path == (
f"/data/data/com.termux/files/home/.cache/hermes-tmp/hermes-snap-{env._session_id}.sh"
)
assert env._cwd_file == (
f"/data/data/com.termux/files/home/.cache/hermes-tmp/hermes-cwd-{env._session_id}.txt"
)
def test_falls_back_to_tempfile_when_tmp_missing(self, monkeypatch):
monkeypatch.delenv("TMPDIR", raising=False)
monkeypatch.delenv("TMP", raising=False)
monkeypatch.delenv("TEMP", raising=False)
with patch("tools.environments.local.os.path.isdir", return_value=False), \
patch("tools.environments.local.os.access", return_value=False), \
patch("tools.environments.local.tempfile.gettempdir", return_value="/cache/tmp"), \
patch.object(LocalEnvironment, "init_session", autospec=True, return_value=None):
env = LocalEnvironment(cwd=".", timeout=10)
assert env.get_temp_dir() == "/cache/tmp"
assert env._snapshot_path == f"/cache/tmp/hermes-snap-{env._session_id}.sh"
assert env._cwd_file == f"/cache/tmp/hermes-cwd-{env._session_id}.txt"

View file

@ -16,6 +16,7 @@ from tools.tool_result_storage import (
STORAGE_DIR,
_build_persisted_message,
_heredoc_marker,
_resolve_storage_dir,
_write_to_sandbox,
enforce_turn_budget,
generate_preview,
@ -115,6 +116,24 @@ class TestWriteToSandbox:
_write_to_sandbox("content", "/tmp/hermes-results/abc.txt", env)
assert env.execute.call_args[1]["timeout"] == 30
def test_uses_parent_dir_of_remote_path(self):
env = MagicMock()
env.execute.return_value = {"output": "", "returncode": 0}
remote_path = "/data/data/com.termux/files/usr/tmp/hermes-results/abc.txt"
_write_to_sandbox("content", remote_path, env)
cmd = env.execute.call_args[0][0]
assert "mkdir -p /data/data/com.termux/files/usr/tmp/hermes-results" in cmd
class TestResolveStorageDir:
def test_defaults_to_storage_dir_without_env(self):
assert _resolve_storage_dir(None) == STORAGE_DIR
def test_uses_env_temp_dir_when_available(self):
env = MagicMock()
env.get_temp_dir.return_value = "/data/data/com.termux/files/usr/tmp"
assert _resolve_storage_dir(env) == "/data/data/com.termux/files/usr/tmp/hermes-results"
# ── _build_persisted_message ──────────────────────────────────────────
@ -341,6 +360,22 @@ class TestMaybePersistToolResult:
)
assert "DISTINCTIVE_START_MARKER" in result
def test_env_temp_dir_changes_persisted_path(self):
env = MagicMock()
env.execute.return_value = {"output": "", "returncode": 0}
env.get_temp_dir.return_value = "/data/data/com.termux/files/usr/tmp"
content = "x" * 60_000
result = maybe_persist_tool_result(
content=content,
tool_name="terminal",
tool_use_id="tc_termux",
env=env,
threshold=30_000,
)
assert "/data/data/com.termux/files/usr/tmp/hermes-results/tc_termux.txt" in result
cmd = env.execute.call_args[0][0]
assert "mkdir -p /data/data/com.termux/files/usr/tmp/hermes-results" in cmd
def test_threshold_zero_forces_persist(self):
env = MagicMock()
env.execute.return_value = {"output": "", "returncode": 0}

View file

@ -226,14 +226,24 @@ class BaseEnvironment(ABC):
# Snapshot creation timeout (override for slow cold-starts).
_snapshot_timeout: int = 30
def get_temp_dir(self) -> str:
"""Return the backend temp directory used for session artifacts.
Most sandboxed backends use ``/tmp`` inside the target environment.
LocalEnvironment overrides this on platforms like Termux where ``/tmp``
may be missing and ``TMPDIR`` is the portable writable location.
"""
return "/tmp"
def __init__(self, cwd: str, timeout: int, env: dict = None):
self.cwd = cwd
self.timeout = timeout
self.env = env or {}
self._session_id = uuid.uuid4().hex[:12]
self._snapshot_path = f"/tmp/hermes-snap-{self._session_id}.sh"
self._cwd_file = f"/tmp/hermes-cwd-{self._session_id}.txt"
temp_dir = self.get_temp_dir().rstrip("/") or "/"
self._snapshot_path = f"{temp_dir}/hermes-snap-{self._session_id}.sh"
self._cwd_file = f"{temp_dir}/hermes-cwd-{self._session_id}.txt"
self._cwd_marker = _cwd_marker(self._session_id)
self._snapshot_ready = False
self._last_sync_time: float | None = (

View file

@ -5,6 +5,7 @@ import platform
import shutil
import signal
import subprocess
import tempfile
from tools.environments.base import BaseEnvironment, _pipe_stdin
@ -209,6 +210,32 @@ class LocalEnvironment(BaseEnvironment):
super().__init__(cwd=cwd or os.getcwd(), timeout=timeout, env=env)
self.init_session()
def get_temp_dir(self) -> str:
"""Return a shell-safe writable temp dir for local execution.
Termux does not provide /tmp by default, but exposes a POSIX TMPDIR.
Prefer POSIX-style env vars when available, keep using /tmp on regular
Unix systems, and only fall back to tempfile.gettempdir() when it also
resolves to a POSIX path.
Check the environment configured for this backend first so callers can
override the temp root explicitly (for example via terminal.env or a
custom TMPDIR), then fall back to the host process environment.
"""
for env_var in ("TMPDIR", "TMP", "TEMP"):
candidate = self.env.get(env_var) or os.environ.get(env_var)
if candidate and candidate.startswith("/"):
return candidate.rstrip("/") or "/"
if os.path.isdir("/tmp") and os.access("/tmp", os.W_OK | os.X_OK):
return "/tmp"
candidate = tempfile.gettempdir()
if candidate.startswith("/"):
return candidate.rstrip("/") or "/"
return "/tmp"
def _run_bash(self, cmd_string: str, *, login: bool = False,
timeout: int = 120,
stdin_data: str | None = None) -> subprocess.Popen:

View file

@ -9,9 +9,11 @@ Defense against context-window overflow operates at three levels:
2. **Per-result persistence** (maybe_persist_tool_result): After a tool
returns, if its output exceeds the tool's registered threshold
(registry.get_max_result_size), the full output is written INTO THE
SANDBOX at /tmp/hermes-results/{tool_use_id}.txt via env.execute().
The in-context content is replaced with a preview + file path reference.
The model can read_file to access the full output on any backend.
SANDBOX temp dir (for example /tmp/hermes-results/{tool_use_id}.txt on
standard Linux, or $TMPDIR/hermes-results/{tool_use_id}.txt on Termux)
via env.execute(). The in-context content is replaced with a preview +
file path reference. The model can read_file to access the full output
on any backend.
3. **Per-turn aggregate budget** (enforce_turn_budget): After all tool
results in a single assistant turn are collected, if the total exceeds
@ -21,6 +23,7 @@ Defense against context-window overflow operates at three levels:
"""
import logging
import os
import uuid
from tools.budget_config import (
@ -37,6 +40,22 @@ HEREDOC_MARKER = "HERMES_PERSIST_EOF"
_BUDGET_TOOL_NAME = "__budget_enforcement__"
def _resolve_storage_dir(env) -> str:
"""Return the best temp-backed storage dir for this environment."""
if env is not None:
get_temp_dir = getattr(env, "get_temp_dir", None)
if callable(get_temp_dir):
try:
temp_dir = get_temp_dir()
except Exception as exc:
logger.debug("Could not resolve env temp dir: %s", exc)
else:
if temp_dir:
temp_dir = temp_dir.rstrip("/") or "/"
return f"{temp_dir}/hermes-results"
return STORAGE_DIR
def generate_preview(content: str, max_chars: int = DEFAULT_PREVIEW_SIZE_CHARS) -> tuple[str, bool]:
"""Truncate at last newline within max_chars. Returns (preview, has_more)."""
if len(content) <= max_chars:
@ -58,8 +77,9 @@ def _heredoc_marker(content: str) -> str:
def _write_to_sandbox(content: str, remote_path: str, env) -> bool:
"""Write content into the sandbox via env.execute(). Returns True on success."""
marker = _heredoc_marker(content)
storage_dir = os.path.dirname(remote_path)
cmd = (
f"mkdir -p {STORAGE_DIR} && cat > {remote_path} << '{marker}'\n"
f"mkdir -p {storage_dir} && cat > {remote_path} << '{marker}'\n"
f"{content}\n"
f"{marker}"
)
@ -125,7 +145,8 @@ def maybe_persist_tool_result(
if len(content) <= effective_threshold:
return content
remote_path = f"{STORAGE_DIR}/{tool_use_id}.txt"
storage_dir = _resolve_storage_dir(env)
remote_path = f"{storage_dir}/{tool_use_id}.txt"
preview, has_more = generate_preview(content, max_chars=config.preview_size)
if env is not None: