diff --git a/tests/tools/test_credential_files.py b/tests/tools/test_credential_files.py index 32b4c7664df..ac2fb53f3a2 100644 --- a/tests/tools/test_credential_files.py +++ b/tests/tools/test_credential_files.py @@ -13,6 +13,7 @@ from tools.credential_files import ( get_skills_directory_mount, iter_cache_files, iter_skills_files, + map_cache_path_to_container, register_credential_file, register_credential_files, ) @@ -423,6 +424,48 @@ class TestCacheDirectoryMounts: assert get_cache_directory_mounts() == [] +class TestMapCachePathToContainer: + """Tests for map_cache_path_to_container() — the backend-agnostic mapper.""" + + def test_maps_path_under_cache_dir(self, tmp_path, monkeypatch): + hermes_home = tmp_path / ".hermes" + img_dir = hermes_home / "cache" / "images" + img_dir.mkdir(parents=True) + host_path = str(img_dir / "generated.png") + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + + assert ( + map_cache_path_to_container(host_path) + == "/root/.hermes/cache/images/generated.png" + ) + + def test_custom_container_base_for_remote_home(self, tmp_path, monkeypatch): + hermes_home = tmp_path / ".hermes" + img_dir = hermes_home / "cache" / "images" + img_dir.mkdir(parents=True) + host_path = str(img_dir / "remote.png") + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + + assert ( + map_cache_path_to_container(host_path, container_base="/home/agent/.hermes") + == "/home/agent/.hermes/cache/images/remote.png" + ) + + def test_returns_none_when_outside_cache_dirs(self, tmp_path, monkeypatch): + hermes_home = tmp_path / ".hermes" + (hermes_home / "cache" / "images").mkdir(parents=True) + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + + assert map_cache_path_to_container(str(tmp_path / "elsewhere.png")) is None + + def test_returns_none_when_no_cache_dirs_exist(self, tmp_path, monkeypatch): + hermes_home = tmp_path / ".hermes" + hermes_home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + + assert map_cache_path_to_container(str(hermes_home / "cache" / "images" / "x.png")) is None + + class TestIterCacheFiles: """Tests for iter_cache_files().""" diff --git a/tools/credential_files.py b/tools/credential_files.py index 381115e0955..5fbd27bd077 100644 --- a/tools/credential_files.py +++ b/tools/credential_files.py @@ -22,9 +22,10 @@ from __future__ import annotations import logging import os +import posixpath from contextvars import ContextVar from pathlib import Path -from typing import Dict, List +from typing import Dict, List, Optional from hermes_cli.config import cfg_get logger = logging.getLogger(__name__) @@ -374,6 +375,30 @@ def get_cache_directory_mounts( return mounts +def map_cache_path_to_container( + host_path: str, + container_base: str = "/root/.hermes", +) -> Optional[str]: + """Map a host cache path to its mounted path under *container_base*. + + Returns the POSIX container path when *host_path* lives under one of the + auto-mounted cache directories, otherwise ``None``. Backend-agnostic: the + caller decides which ``container_base`` applies (Docker ``/root/.hermes``, + SSH ``/.hermes``, etc.) and whether translation is wanted. + Always joins with ``posixpath`` because container/remote paths are POSIX + regardless of the host OS. + """ + path = Path(host_path) + for mount in get_cache_directory_mounts(container_base=container_base): + host_dir = Path(mount["host_path"]) + try: + rel = path.relative_to(host_dir) + except ValueError: + continue + return posixpath.join(mount["container_path"], rel.as_posix()) + return None + + def to_agent_visible_cache_path( host_path: str, container_base: str = "/root/.hermes", @@ -391,15 +416,8 @@ def to_agent_visible_cache_path( if os.environ.get("TERMINAL_ENV", "local") != "docker": return host_path - path = Path(host_path) - for mount in get_cache_directory_mounts(container_base=container_base): - host_dir = Path(mount["host_path"]) - try: - rel = path.relative_to(host_dir) - return str(Path(mount["container_path"]) / rel) - except ValueError: - continue - return host_path + mapped = map_cache_path_to_container(host_path, container_base=container_base) + return mapped if mapped is not None else host_path def iter_cache_files( diff --git a/tools/image_generation_tool.py b/tools/image_generation_tool.py index 7e2080a7a16..d7eeb30d175 100644 --- a/tools/image_generation_tool.py +++ b/tools/image_generation_tool.py @@ -23,11 +23,9 @@ update when it's noticed. import json import logging import os -import posixpath import datetime import threading import uuid -from pathlib import Path from typing import Any, Dict, Optional # fal_client is imported lazily — see _load_fal_client(). Pulling it @@ -631,6 +629,10 @@ def _active_terminal_env(task_id: str | None): def _agent_cache_base_for_env(env: Any) -> str | None: if env is not None: + # Forward-looking optional override: an environment may expose its own + # agent-visible cache root via this callable. No backend defines it yet + # — it's an extension hook, not a typo. The getattr/callable guards make + # it a safe no-op until a producer exists. explicit = getattr(env, "agent_visible_cache_base", None) if callable(explicit): try: @@ -669,16 +671,9 @@ def _agent_visible_cache_path(host_path: str, env: Any) -> str | None: return None try: - from tools.credential_files import get_cache_directory_mounts + from tools.credential_files import map_cache_path_to_container - path = Path(host_path) - for mount in get_cache_directory_mounts(container_base=cache_base): - host_dir = Path(mount["host_path"]) - try: - rel = path.relative_to(host_dir) - except ValueError: - continue - return posixpath.join(mount["container_path"], rel.as_posix()) + return map_cache_path_to_container(host_path, container_base=cache_base) except Exception as exc: # noqa: BLE001 logger.debug("Could not translate image cache path for backend: %s", exc) return None