fix(sprites): make cache artifacts and skill helpers usable on Toolbox

agent/skill_commands.py

@@ -8,7 +8,7 @@
 import logging
 import os
 import re
-from pathlib import Path
+from pathlib import Path, PurePosixPath
 from typing import Any, Dict, Optional
 
 from hermes_constants import display_hermes_home
@@ -296,19 +296,23 @@ def _build_skill_message(
     # Done before anything else so downstream blocks (setup notes,
     # supporting-file hints) see the expanded content.
     skills_cfg = _load_skills_config()
-    if skills_cfg.get("template_vars", True):
-        content = _substitute_template_vars(content, skill_dir, session_id)
     if skills_cfg.get("inline_shell", False):
         timeout = int(skills_cfg.get("inline_shell_timeout", 10) or 10)
-        content = _expand_inline_shell(content, skill_dir, timeout)
+        content = _expand_inline_shell(content, skill_dir, timeout, session_id, skills_cfg.get("template_vars", True))
+    if skills_cfg.get("template_vars", True):
+        content = _substitute_template_vars(content, skill_dir, session_id)
+
+    from agent.skill_path_mapping import map_skill_dir_for_backend
 
+    mapped_dir = map_skill_dir_for_backend(skill_dir, task_id=session_id) if skill_dir else None
+    hint_dir = PurePosixPath(mapped_dir) if mapped_dir and mapped_dir != str(skill_dir) else skill_dir
     parts = [activation_note, "", content.strip()]
 
     # ── Inject the absolute skill directory so the agent can reference
     #    bundled scripts without an extra skill_view() round-trip. ──
     if skill_dir:
         parts.append("")
-        parts.append(f"[Skill directory: {skill_dir}]")
+        parts.append(f"[Skill directory: {hint_dir}]")
         parts.append(
             "Resolve any relative paths in this skill (e.g. `scripts/foo.js`, "
             "`templates/config.yaml`) against that directory, then run them "
@@ -364,11 +368,11 @@ def _build_skill_message(
         parts.append("")
         parts.append("[This skill has supporting files:]")
         for sf in supporting:
-            parts.append(f"- {sf}  ->  {skill_dir / sf}")
+            parts.append(f"- {sf}  ->  {hint_dir / sf}")
         parts.append(
             f'\nLoad any of these with skill_view(name="{skill_view_target}", '
             f'file_path="<path>"), or run scripts directly by absolute path '
-            f"(e.g. `node {skill_dir}/scripts/foo.js`)."
+            f"(e.g. `node {hint_dir}/scripts/foo.js`)."
         )
 
     if user_instruction:

agent/skill_path_mapping.py

@@ -0,0 +1,171 @@
+"""Map host skill directory paths to backend-visible paths.
+
+When the active terminal backend is remote (Docker, SSH, Daytona, Singularity,
+Modal), the skills tree lives at a different filesystem location inside the
+sandbox than on the host.  Skill content that references the skill directory
+(``${HERMES_SKILL_DIR}``, ``[Skill directory: ...]``, supporting-file hints)
+must use the backend-visible path, or the agent will try to run bundled
+scripts via paths that do not exist in the sandbox (hermes-agent#41541,
+#73842).
+
+The authoritative mount layout is computed by
+``tools.credential_files.get_skills_directory_layout()``; this module consumes
+it with a longest-prefix match and falls back to the host path whenever the
+backend is local or unknown, so behavior on local backends is unchanged.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from pathlib import Path, PurePosixPath
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# Container backends whose skills mount root is /root/.hermes inside the
+# sandbox.  Mirrors the class names in tools/environments/*.py.
+_CONTAINER_ENV_CLASSES = {
+    "DockerEnvironment",
+    "SingularityEnvironment",
+    "ModalEnvironment",
+    "ManagedModalEnvironment",
+}
+
+# TERMINAL_ENV values that map to the /root/.hermes container layout even
+# before a live environment object exists (used at first skill render).
+_CONTAINER_BACKEND_NAMES = {"docker", "singularity", "modal"}
+
+# Remote backends whose .hermes root is only known from the live environment
+# (SSH/Daytona resolve a remote home at connect time).
+_REMOTE_BACKEND_NAMES = {"ssh", "daytona"}
+
+
+def _active_terminal_env(task_id: str | None) -> Any:
+    """Return the live terminal environment for *task_id*, or None."""
+    if not task_id:
+        return None
+    try:
+        from tools.terminal_tool import get_active_env
+
+        return get_active_env(task_id)
+    except Exception:
+        logger.debug("Could not resolve active terminal env", exc_info=True)
+        return None
+
+
+def _backend_name() -> str:
+    return str(os.getenv("TERMINAL_ENV", "local")).strip().lower() or "local"
+
+
+def _hermes_base_for_env(env: Any, backend_name: str) -> str | None:
+    """Resolve the backend-visible ``.hermes`` root, or None if unknown.
+
+    None means "the host path is the backend path" (local/unknown backend) or
+    the backend root cannot be determined without a live environment.
+    """
+    if env is not None:
+        remote_home = getattr(env, "_remote_home", None)
+        if remote_home:
+            return f"{str(remote_home).rstrip('/')}/.hermes"
+        if type(env).__name__ in _CONTAINER_ENV_CLASSES:
+            return "/root/.hermes"
+        return None
+    if backend_name in _CONTAINER_BACKEND_NAMES:
+        return "/root/.hermes"
+    return None
+
+
+def map_skill_dir_for_backend(
+    host_skill_dir: Path | str | None,
+    task_id: str | None = None,
+) -> str:
+    """Translate *host_skill_dir* to the path the agent sees on the backend.
+
+    Longest-prefix-matches the host path against the existing skills mount
+    layout (``get_skills_directory_layout``) and returns the corresponding
+    backend-visible path (POSIX form, since container/remote paths are
+    POSIX).  Falls back to the host path unchanged when:
+
+    - the backend is local or unknown (TERMINAL_ENV unset / "local"),
+    - the backend root cannot be determined without a live environment
+      (SSH/Daytona before first connect),
+    - the directory is not under any known skills mount, or
+    - the mount layout cannot be resolved.
+    """
+    if host_skill_dir is None:
+        return ""
+    host = str(host_skill_dir)
+    if _backend_name() == "sprites":
+        return _map_sprites_skill_dir(host)
+    base = _hermes_base_for_env(_active_terminal_env(task_id), _backend_name())
+    if not base:
+        return host
+    try:
+        from tools.credential_files import get_skills_directory_layout
+
+        mounts = get_skills_directory_layout(container_base=base)
+    except Exception:
+        logger.debug("Could not resolve skills directory mount layout", exc_info=True)
+        return host
+    if not mounts:
+        return host
+
+    # Longest-prefix match against mount host paths.  Normalize separators so
+    # Windows hosts (backslash paths) match against POSIX container prefixes.
+    # On Windows, filesystems are case-insensitive, so comparisons are
+    # lowercased there; on POSIX hosts the match stays case-sensitive.
+    host_norm = host.replace("\\", "/")
+    case_fold = os.name == "nt"
+    best_prefix: str | None = None
+    best_container: str | None = None
+    for m in mounts:
+        # Match against the canonical source path AND the actual mount
+        # source: when symlinks are present the mount host_path is a
+        # sanitized copy while agent-visible skill dirs live under the
+        # canonical skills tree (source_path).
+        for key in ("source_path", "host_path"):
+            candidate = m.get(key)
+            if not candidate:
+                continue
+            prefix = str(candidate).rstrip("/").replace("\\", "/")
+            match_norm = host_norm if not case_fold else host_norm.lower()
+            prefix_norm = prefix if not case_fold else prefix.lower()
+            if match_norm == prefix_norm or match_norm.startswith(prefix_norm + "/"):
+                if best_prefix is None or len(prefix) > len(best_prefix):
+                    best_prefix = prefix
+                    best_container = m["container_path"]
+    if best_container is None:
+        return host
+
+    rel = host_norm[len(best_prefix):].lstrip("/")
+    if not rel:
+        return best_container
+    return f"{best_container.rstrip('/')}/{rel}"
+
+
+def _map_sprites_skill_dir(host: str) -> str:
+    """Use the same roots as Sprites' secret-free skills sync (no host mounts)."""
+    from agent.skill_utils import get_external_skills_dirs
+    from tools.credential_files import _resolve_hermes_home
+
+    roots = [(_resolve_hermes_home() / "skills", "/skills")]
+    roots.extend((root, f"/skills/external_skills/{index}")
+                 for index, root in enumerate(get_external_skills_dirs()))
+    path = Path(host).absolute()
+    for root, target in sorted(roots, key=lambda item: len(item[0].parts), reverse=True):
+        try:
+            relative = path.relative_to(root.absolute())
+        except ValueError:
+            continue
+        # The sync skips symlinks below each catalog root. Never advertise
+        # a path to bytes that are intentionally excluded from the projection.
+        current = root
+        if ".." in relative.parts:
+            return host
+        for part in relative.parts:
+            current = current / part
+            if current.is_symlink():
+                return host
+        return str(PurePosixPath(target) / relative.as_posix())
+    return host

agent/skill_preprocessing.py

@@ -40,6 +40,8 @@ def substitute_template_vars(
     content: str,
     skill_dir: Path | None,
     session_id: str | None,
+    *,
+    runtime_paths: bool = True,
 ) -> str:
     """Replace ${HERMES_SKILL_DIR} / ${HERMES_SESSION_ID} in skill content.
 
@@ -49,7 +51,11 @@ def substitute_template_vars(
     if not content:
         return content
 
+    from agent.skill_path_mapping import map_skill_dir_for_backend
+
     skill_dir_str = str(skill_dir) if skill_dir else None
+    if skill_dir and runtime_paths:
+        skill_dir_str = map_skill_dir_for_backend(skill_dir, task_id=session_id)
 
     def _replace(match: re.Match) -> str:
         token = match.group(1)
@@ -107,6 +113,8 @@ def expand_inline_shell(
     content: str,
     skill_dir: Path | None,
     timeout: int,
+    session_id: str | None = None,
+    template_vars: bool = False,
 ) -> str:
     """Replace every !`cmd` snippet in ``content`` with its stdout.
 
@@ -120,6 +128,10 @@ def _replace(match: re.Match) -> str:
         cmd = match.group(1).strip()
         if not cmd:
             return ""
+        if template_vars:
+            # Inline preprocessing still runs on the host. Runtime path
+            # translation applies only to instructions handed to the agent.
+            cmd = substitute_template_vars(cmd, skill_dir, session_id, runtime_paths=False)
         return run_inline_shell(cmd, skill_dir, timeout)
 
     return _INLINE_SHELL_RE.sub(_replace, content)
@@ -136,9 +148,9 @@ def preprocess_skill_content(
         return content
 
     cfg = skills_cfg if isinstance(skills_cfg, dict) else load_skills_config()
-    if cfg.get("template_vars", True):
-        content = substitute_template_vars(content, skill_dir, session_id)
     if cfg.get("inline_shell", False):
         timeout = int(cfg.get("inline_shell_timeout", 10) or 10)
-        content = expand_inline_shell(content, skill_dir, timeout)
+        content = expand_inline_shell(content, skill_dir, timeout, session_id, cfg.get("template_vars", True))
+    if cfg.get("template_vars", True):
+        content = substitute_template_vars(content, skill_dir, session_id)
     return content

agent/skill_utils.py

@@ -749,6 +749,21 @@ def _resolve_dotpath(config: Dict[str, Any], dotted_key: str):
     return current
 
 
+_HOME_VAR_RE = re.compile(r"\$(?:\{HOME\}|HOME)(?=$|[/\\])")
+
+
+def _expand_skill_config_path(value: str) -> str:
+    """Expand skill defaults against tool HOME, not the gateway account."""
+    from hermes_constants import get_subprocess_home
+
+    tool_home = "/home" if os.getenv("TERMINAL_ENV", "").strip().lower() == "sprites" else get_subprocess_home()
+    if tool_home:
+        if value == "~" or value.startswith(("~/", "~\\")):
+            value = tool_home + value[1:]
+        value = _HOME_VAR_RE.sub(lambda _match: tool_home, value)
+    return os.path.expanduser(os.path.expandvars(value))
+
+
 def resolve_skill_config_values(
     config_vars: List[Dict[str, Any]],
 ) -> Dict[str, Any]:
@@ -771,8 +786,8 @@ def resolve_skill_config_values(
             value = var.get("default", "")
 
         # Expand ~ in path-like values
-        if isinstance(value, str) and ("~" in value or "${" in value):
-            value = os.path.expanduser(os.path.expandvars(value))
+        if isinstance(value, str) and ("~" in value or "$" in value):
+            value = _expand_skill_config_path(value)
 
         resolved[logical_key] = value
 

gateway/platforms/base.py

@@ -1826,7 +1826,7 @@ def cache_media_bytes(
     ``application/octet-stream``); only images that fail validation
     (``cache_image_from_bytes`` raises ValueError) return None.
     """
-    from tools.credential_files import to_agent_visible_cache_path
+    from tools.credential_files import publish_cache_path
 
     ext = _resolve_media_ext(filename, mime_type)
     mime = (mime_type or "").lower()
@@ -1847,18 +1847,18 @@ def cache_media_bytes(
         except ValueError:
             return None
         out_mime = mime if mime.startswith("image/") else SUPPORTED_IMAGE_DOCUMENT_TYPES.get(img_ext, "image/jpeg")
-        return CachedMedia(to_agent_visible_cache_path(path), out_mime, "image", display)
+        return CachedMedia(publish_cache_path(path), out_mime, "image", display)
 
     if is_video:
         vid_ext = ext if ext in SUPPORTED_VIDEO_TYPES else ".mp4"
         path = cache_video_from_bytes(data, ext=vid_ext)
-        return CachedMedia(to_agent_visible_cache_path(path), SUPPORTED_VIDEO_TYPES.get(vid_ext, "video/mp4"), "video", display)
+        return CachedMedia(publish_cache_path(path), SUPPORTED_VIDEO_TYPES.get(vid_ext, "video/mp4"), "video", display)
 
     if is_audio:
         aud_ext = ext if ext in {".ogg", ".mp3", ".wav", ".m4a", ".opus", ".flac"} else ".ogg"
         path = cache_audio_from_bytes(data, ext=aud_ext)
         out_mime = mime if mime.startswith("audio/") else f"audio/{aud_ext.lstrip('.')}"
-        return CachedMedia(to_agent_visible_cache_path(path), out_mime, "audio", display)
+        return CachedMedia(publish_cache_path(path), out_mime, "audio", display)
 
     # Any other file type is cached and surfaced to the agent as a local path
     # so it can be inspected with terminal / read_file / etc. Authorization to
@@ -1873,7 +1873,7 @@ def cache_media_bytes(
         out_mime = SUPPORTED_DOCUMENT_TYPES[ext]
     else:
         out_mime = mime if mime else "application/octet-stream"
-    return CachedMedia(to_agent_visible_cache_path(path), out_mime, "document", display or fallback_name)
+    return CachedMedia(publish_cache_path(path), out_mime, "document", display or fallback_name)
 
 
 class MessageType(Enum):

gateway/run.py

@@ -12858,7 +12858,7 @@ async def _prepare_inbound_message_text(
                 # language. The hardcoded send has therefore been removed.
 
         if audio_file_paths:
-            from tools.credential_files import to_agent_visible_cache_path as _to_agent_path
+            from tools.credential_files import publish_cache_path as _to_agent_path
             for _apath in audio_file_paths:
                 _basename = os.path.basename(_apath)
                 _parts = _basename.split("_", 2)
@@ -12877,7 +12877,7 @@ async def _prepare_inbound_message_text(
                 message_text = f"{_note}\n\n{message_text}"
 
         if video_paths:
-            from tools.credential_files import to_agent_visible_cache_path as _to_agent_path
+            from tools.credential_files import publish_cache_path as _to_agent_path
             for _vpath in video_paths:
                 _basename = os.path.basename(_vpath)
                 _parts = _basename.split("_", 2)
@@ -12897,7 +12897,7 @@ async def _prepare_inbound_message_text(
 
         if event.media_urls:
             import mimetypes as _mimetypes
-            from tools.credential_files import to_agent_visible_cache_path
+            from tools.credential_files import publish_cache_path
 
             _TEXT_EXTENSIONS = {".txt", ".md", ".csv", ".log", ".json", ".xml", ".yaml", ".yml", ".toml", ".ini", ".cfg"}
             for i, path in enumerate(event.media_urls):
@@ -12934,10 +12934,9 @@ async def _prepare_inbound_message_text(
                 display_name = parts[2] if len(parts) >= 3 else basename
                 display_name = re.sub(r'[^\w.\- ]', '_', display_name)
 
-                # Translate host cache path to in-container path if running under Docker backend.
-                # This ensures the agent receives a path it can open inside its sandbox, as the
-                # cache directories are auto-mounted at /root/.hermes/cache/* by get_cache_directory_mounts().
-                agent_path = to_agent_visible_cache_path(path)
+                # Publish before advertising the path: copying backends need
+                # the bytes on the remote filesystem, while Docker uses mounts.
+                agent_path = publish_cache_path(path)
 
                 context_note = _build_document_context_note(display_name, agent_path, mtype)
                 message_text = f"{context_note}\n\n{message_text}"