refactor(image_gen): delegate cache-path mapping to shared helper

Follow-up on the backend-visible artifact-path fix.

- Extract the cache-mount iteration loop into a reusable, backend-agnostic
  credential_files.map_cache_path_to_container(host_path, container_base) that
  returns the POSIX container path or None. to_agent_visible_cache_path() now
  delegates to it (keeping its Docker-only gate), and image_generation_tool's
  _agent_visible_cache_path() delegates to it too — eliminating the duplicated
  loop and the divergent path-join (posixpath vs Path) between the two.
- Drop the now-unused posixpath/Path imports from image_generation_tool.py.
- Document the agent_visible_cache_base getattr probe as a forward-looking
  optional hook (no producer yet) so it doesn't read as a typo'd attribute.
- Add unit tests for map_cache_path_to_container.
This commit is contained in:
kshitijk4poor 2026-06-07 01:43:33 +05:30 committed by kshitij
parent 7c4aa3e4da
commit c79e3fd0ba
3 changed files with 77 additions and 21 deletions

View file

@ -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()."""

View file

@ -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 ``<remote_home>/.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(

View file

@ -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