diff --git a/tools/arg_coercion.py b/tools/arg_coercion.py
index d3406159bb..3287565def 100644
--- a/tools/arg_coercion.py
+++ b/tools/arg_coercion.py
@@ -2,8 +2,8 @@
 
 Models emit "42" for integers, "true" for booleans, JSON-encoded strings for
 arrays/objects (also nested inside containers), and bare scalars where an array
-is expected. Coercion is schema-guided and conservative: originals are kept
-whenever a repair is not unambiguous.
+is expected (wrapped in a one-element list). Coercion is schema-guided and
+conservative: originals are kept whenever a repair is not unambiguous.
 """
 
 import json
@@ -18,20 +18,12 @@ logger = logging.getLogger("model_tools")
 
 
 def coerce_tool_args(tool_name: str, args: Dict[str, Any]) -> Dict[str, Any]:
-    """Coerce string-typed args to their JSON-Schema types; originals kept on failure.
-
-    Models emit "42" for integers, "true" for booleans, JSON-encoded strings for
-    arrays/objects (also nested inside containers), and bare scalars where an
-    array is expected (wrapped in a one-element list).
-    """
+    """Coerce string-typed args to their JSON-Schema types; originals kept on failure."""
     if not args or not isinstance(args, dict):
         return args
 
     schema = registry.get_schema(tool_name)
-    if not schema:
-        return args
-
-    properties = (schema.get("parameters") or {}).get("properties")
+    properties = ((schema or {}).get("parameters") or {}).get("properties")
     if not properties:
         return args
 
@@ -48,43 +40,33 @@ def coerce_tool_args(tool_name: str, args: Dict[str, Any]) -> Dict[str, Any]:
         if not prop_schema:
             continue
         expected = prop_schema.get("type")
+        is_container = isinstance(value, (list, tuple))
 
         # Bare non-list value for an array schema. Strings go through
         # _coerce_value first so a JSON-encoded array is parsed and a nullable
         # "null" becomes None (not ["null"]). None itself is preserved: the tool's
         # own default handling decides between "omit" and "empty list".
-        if expected == "array" and value is not None and not isinstance(value, (list, tuple)):
+        if expected == "array" and value is not None and not is_container:
             if isinstance(value, str):
                 coerced = _coerce_value(value, expected, schema=prop_schema)
                 if coerced is not value:
                     args[key] = coerced
                     continue
                 if value.strip().startswith("["):
-                    logger.warning(
-                        "coerce_tool_args: %s.%s looks like a JSON array string "
-                        "but could not be parsed — model may have emitted a "
-                        "JSON-encoded string instead of a native array. "
-                        "Falling back to single-element list.",
-                        tool_name, key,
-                    )
+                    logger.warning("coerce_tool_args: %s.%s looks like a JSON array string "
+                                   "but could not be parsed — model may have emitted a "
+                                   "JSON-encoded string instead of a native array. "
+                                   "Falling back to single-element list.", tool_name, key)
                 args[key] = [value]
-                logger.info(
-                    "coerce_tool_args: wrapped bare string in list for %s.%s",
-                    tool_name, key,
-                )
+                logger.info("coerce_tool_args: wrapped bare string in list for %s.%s", tool_name, key)
                 continue
             args[key] = [value]
-            logger.info(
-                "coerce_tool_args: wrapped bare %s in list for %s.%s",
-                type(value).__name__, tool_name, key,
-            )
+            logger.info("coerce_tool_args: wrapped bare %s in list for %s.%s", type(value).__name__, tool_name, key)
             continue
 
         if not isinstance(value, str):
             # Native container: still normalize JSON-encoded elements/sub-fields.
-            if (expected == "array" and isinstance(value, (list, tuple))) or (
-                expected == "object" and isinstance(value, dict)
-            ):
+            if (expected == "array" and is_container) or (expected == "object" and isinstance(value, dict)):
                 args[key] = _normalize_json_strings_for_schema(value, prop_schema)
             continue
         if not expected and not _schema_allows_null(prop_schema):
@@ -105,13 +87,8 @@ def _schema_accepts_kind(schema: Any, kind: str) -> bool:
     t = schema.get("type")
     if t == kind or (isinstance(t, list) and kind in t):
         return True
-    for union_key in ("anyOf", "oneOf", "allOf"):
-        branches = schema.get(union_key)
-        if isinstance(branches, list) and any(
-            _schema_accepts_kind(b, kind) for b in branches
-        ):
-            return True
-    return False
+    return any(isinstance(branches := schema.get(union_key), list) and any(_schema_accepts_kind(b, kind) for b in branches)
+               for union_key in ("anyOf", "oneOf", "allOf"))
 
 
 def _normalize_json_strings_for_schema(value: Any, schema: Any) -> Any:
@@ -128,46 +105,32 @@ def _normalize_json_strings_for_schema(value: Any, schema: Any) -> Any:
         trimmed = value.strip()
         expects_array = _schema_accepts_kind(schema, "array")
         expects_object = _schema_accepts_kind(schema, "object")
-        if (expects_array and trimmed.startswith("[")) or (
-            expects_object and trimmed.startswith("{")
-        ):
-            try:
-                parsed = json.loads(trimmed)
-            except (ValueError, TypeError):
-                return value
-            if (isinstance(parsed, list) and expects_array) or (isinstance(parsed, dict) and expects_object):
-                value = parsed
-            else:
-                return value
-        else:
+        if not ((expects_array and trimmed.startswith("[")) or (expects_object and trimmed.startswith("{"))):
             return value
+        try:
+            parsed = json.loads(trimmed)
+        except (ValueError, TypeError):
+            return value
+        if not ((isinstance(parsed, list) and expects_array) or (isinstance(parsed, dict) and expects_object)):
+            return value
+        value = parsed
 
     if isinstance(value, list):
         items_schema = schema.get("items")
         if not isinstance(items_schema, dict):
             return value
-        changed = False
-        out = []
-        for item in value:
-            nxt = _normalize_json_strings_for_schema(item, items_schema)
-            changed = changed or (nxt is not item)
-            out.append(nxt)
-        return out if changed else value
+        out = [_normalize_json_strings_for_schema(item, items_schema) for item in value]
+        return out if any(n is not o for n, o in zip(out, value)) else value
 
     if isinstance(value, dict):
         props = schema.get("properties")
         if not isinstance(props, dict):
             return value
-        changed = False
         out = dict(value)
         for k, prop_schema in props.items():
-            if k not in value or not isinstance(prop_schema, dict):
-                continue
-            nxt = _normalize_json_strings_for_schema(value[k], prop_schema)
-            if nxt is not value[k]:
-                out[k] = nxt
-                changed = True
-        return out if changed else value
+            if k in value and isinstance(prop_schema, dict):
+                out[k] = _normalize_json_strings_for_schema(value[k], prop_schema)
+        return out if any(out[k] is not v for k, v in value.items()) else value
 
     return value
 
@@ -178,23 +141,12 @@ def _coerce_value(value: str, expected_type, schema: dict | None = None):
         return None
 
     if isinstance(expected_type, list):
-        for t in expected_type:
-            result = _coerce_value(value, t, schema=schema)
-            if result is not value:
-                return result
-        return value
+        return next((r for t in expected_type if (r := _coerce_value(value, t, schema=schema)) is not value), value)
 
-    if expected_type in {"integer", "number"}:
-        return _coerce_number(value, integer_only=(expected_type == "integer"))
-    if expected_type == "boolean":
-        return _coerce_boolean(value)
-    if expected_type == "array":
-        return _coerce_json(value, list)
-    if expected_type == "object":
-        return _coerce_json(value, dict)
-    if expected_type == "null" and value.strip().lower() == "null":
-        return None
-    return value
+    coercer = _SCALAR_COERCERS.get(expected_type)
+    if coercer is not None:
+        return coercer(value)
+    return None if expected_type == "null" and value.strip().lower() == "null" else value
 
 
 def _schema_allows_null(schema: dict | None) -> bool:
@@ -206,37 +158,24 @@ def _schema_allows_null(schema: dict | None) -> bool:
         return True
     if schema.get("nullable") is True:
         return True
-    for union_key in ("anyOf", "oneOf"):
-        variants = schema.get(union_key)
-        if isinstance(variants, list) and any(
-            isinstance(v, dict) and v.get("type") == "null" for v in variants
-        ):
-            return True
-    return False
+    return any(isinstance(variants := schema.get(union_key), list)
+               and any(isinstance(v, dict) and v.get("type") == "null" for v in variants)
+               for union_key in ("anyOf", "oneOf"))
 
 
 def _coerce_json(value: str, expected_python_type: type):
     """json.loads *value* when the schema expects array/object; original string on mismatch."""
+    name = expected_python_type.__name__
     try:
         parsed = json.loads(value)
     except (ValueError, TypeError) as exc:
-        logger.warning(
-            "coerce_tool_args: failed to parse string as JSON for expected type %s: %s",
-            expected_python_type.__name__,
-            exc,
-        )
+        logger.warning("coerce_tool_args: failed to parse string as JSON for expected type %s: %s", name, exc)
         return value
     if isinstance(parsed, expected_python_type):
-        logger.debug(
-            "coerce_tool_args: coerced string to %s via json.loads",
-            expected_python_type.__name__,
-        )
+        logger.debug("coerce_tool_args: coerced string to %s via json.loads", name)
         return parsed
-    logger.warning(
-        "coerce_tool_args: JSON-parsed value is %s, expected %s — skipping coercion",
-        type(parsed).__name__,
-        expected_python_type.__name__,
-    )
+    logger.warning("coerce_tool_args: JSON-parsed value is %s, expected %s — skipping coercion",
+                   type(parsed).__name__, name)
     return value
 
 
@@ -246,20 +185,21 @@ def _coerce_number(value: str, integer_only: bool = False):
         f = float(value)
     except (ValueError, OverflowError):
         return value
-    if f != f or f == float("inf") or f == float("-inf"):
+    if f != f or f in (float("inf"), float("-inf")):
         return value  # not JSON-serializable
-    if f == int(f):
-        return int(f)
-    if integer_only:
-        return value
-    return f
+    return int(f) if f == int(f) else value if integer_only else f
 
 
 def _coerce_boolean(value: str):
     """Parse "true"/"false" (case-insensitive); original string otherwise."""
-    low = value.strip().lower()
-    if low == "true":
-        return True
-    if low == "false":
-        return False
-    return value
+    return {"true": True, "false": False}.get(value.strip().lower(), value)
+
+
+# JSON-Schema scalar/container type -> coercer; "null" and unions are handled in _coerce_value.
+_SCALAR_COERCERS = {
+    "integer": lambda v: _coerce_number(v, integer_only=True),
+    "number": _coerce_number,
+    "boolean": _coerce_boolean,
+    "array": lambda v: _coerce_json(v, list),
+    "object": lambda v: _coerce_json(v, dict),
+}
diff --git a/tools/blueprints.py b/tools/blueprints.py
index 526c5a93e5..f111c5203a 100644
--- a/tools/blueprints.py
+++ b/tools/blueprints.py
@@ -1,21 +1,10 @@
 """Blueprints: shareable plain-language automations layered on skills + cron.
 
-A "blueprint" is NOT a new object type. It is an ordinary skill (a SKILL.md the
-agent loads) that additionally declares an automation schedule in its
-frontmatter:
-
-    metadata:
-      hermes:
-        blueprint:
-          schedule: "0 9 * * *"     # presence of `blueprint:` marks it runnable
-          deliver: origin            # optional (default "origin")
-          prompt: "..."              # optional task instruction for the run
-          no_agent: false            # optional
-
-Because a blueprint is just a skill it rides the whole skills-hub pipeline
-(search, scan, install, provenance, publish) for free; this module is only the
-bridge from that frontmatter to the cron ``create_job()`` API, plus the inverse
-(``export_blueprint``) that renders a cron job back into a shareable SKILL.md.
+A blueprint is NOT a new object type: it is an ordinary skill whose frontmatter declares
+``metadata.hermes.blueprint`` (``schedule`` required; optional ``deliver`` [default "origin"],
+``prompt``, ``no_agent``, ``model``, ``provider``, ``enabled_toolsets``), so it rides the whole
+skills-hub pipeline for free. This module only bridges that block to cron ``create_job()``,
+plus the inverse (``export_blueprint``) back to a SKILL.md.
 """
 
 from __future__ import annotations
@@ -27,16 +16,8 @@ from typing import Any, Dict, List, Optional
 
 logger = logging.getLogger(__name__)
 
-__all__ = [
-    "BlueprintSpec",
-    "parse_blueprint",
-    "blueprint_spec_for_installed",
-    "blueprint_to_job_spec",
-    "create_blueprint_job",
-    "register_blueprint_suggestion",
-    "export_blueprint",
-    "BlueprintError",
-]
+__all__ = ["BlueprintSpec", "parse_blueprint", "blueprint_spec_for_installed", "blueprint_to_job_spec",
+           "create_blueprint_job", "register_blueprint_suggestion", "export_blueprint", "BlueprintError"]
 
 
 class BlueprintError(ValueError):
@@ -63,18 +44,12 @@ def _split_frontmatter(text: str) -> Optional[Dict[str, Any]]:
     if not isinstance(text, str):
         return None
     stripped = text.lstrip("\ufeff").lstrip()  # BOM is not whitespace; strip explicitly
-    if not stripped.startswith("---"):
+    if not stripped.startswith("---") or (end := stripped.find("\n---", 3)) == -1:
         return None
-    # Find the closing fence after the opening one.
-    after_open = stripped[3:]
-    end = after_open.find("\n---")
-    if end == -1:
-        return None
-    fm_text = after_open[:end]
     try:
         import yaml
 
-        data = yaml.safe_load(fm_text)
+        data = yaml.safe_load(stripped[3:end])
     except Exception as e:  # pragma: no cover - malformed YAML
         logger.debug("blueprint: frontmatter YAML parse failed: %s", e)
         return None
@@ -84,16 +59,14 @@ def _split_frontmatter(text: str) -> Optional[Dict[str, Any]]:
 def parse_blueprint(skill_md_text: str) -> Optional[BlueprintSpec]:
     """Extract a BlueprintSpec from a SKILL.md string, or None if not a blueprint.
 
-    A skill is a blueprint iff ``metadata.hermes.blueprint`` is a mapping containing
-    a non-empty ``schedule``. Raises BlueprintError if the block exists but is
+    A skill is a blueprint iff ``metadata.hermes.blueprint`` is a mapping with a
+    non-empty ``schedule``. Raises BlueprintError if the block exists but is
     structurally invalid (so a typo surfaces instead of silently no-op'ing).
     """
     fm = _split_frontmatter(skill_md_text)
     if not fm:
         return None
 
-    name = str(fm.get("name", "")).strip()
-
     meta = fm.get("metadata")
     hermes = meta.get("hermes") if isinstance(meta, dict) else None
     blueprint = hermes.get("blueprint") if isinstance(hermes, dict) else None
@@ -106,16 +79,13 @@ def parse_blueprint(skill_md_text: str) -> Optional[BlueprintSpec]:
     if not schedule:
         raise BlueprintError("blueprint.schedule is required and must be non-empty")
 
-    prompt = blueprint.get("prompt")
-    model = blueprint.get("model")
-    provider = blueprint.get("provider")
+    prompt, model, provider = blueprint.get("prompt"), blueprint.get("model"), blueprint.get("provider")
     toolsets = blueprint.get("enabled_toolsets")
     if toolsets is not None and not isinstance(toolsets, list):
         raise BlueprintError("blueprint.enabled_toolsets must be a list when present")
 
     return BlueprintSpec(
-        skill_name=name,
-        schedule=schedule,
+        skill_name=str(fm.get("name", "")).strip(), schedule=schedule,
         deliver=str(blueprint.get("deliver", "origin")).strip() or "origin",
         prompt=str(prompt) if prompt is not None else None,
         no_agent=bool(blueprint.get("no_agent", False)),
@@ -133,48 +103,31 @@ def blueprint_spec_for_installed(skill_name: str) -> Optional[BlueprintSpec]:
         from tools.skills_hub import SKILLS_DIR
     except Exception:  # pragma: no cover - import guard
         return None
-
     # Skills live at skills/<category>/<name>/SKILL.md or skills/<name>/SKILL.md.
     for path in Path(SKILLS_DIR).glob(f"**/{skill_name}/SKILL.md"):
         try:
-            text = path.read_text(encoding="utf-8")
+            spec = parse_blueprint(path.read_text(encoding="utf-8"))
         except OSError:
             continue
-        spec = parse_blueprint(text)
         if spec is not None:
-            # Prefer the frontmatter name, fall back to the directory name.
-            if not spec.skill_name:
-                spec.skill_name = skill_name
+            spec.skill_name = spec.skill_name or skill_name  # frontmatter name wins over dir name
             return spec
     return None
 
 
-def blueprint_to_job_spec(
-    spec: BlueprintSpec,
-    *,
-    name: Optional[str] = None,
-) -> Dict[str, Any]:
+def blueprint_to_job_spec(spec: BlueprintSpec, *, name: Optional[str] = None) -> Dict[str, Any]:
     """``cron.jobs.create_job`` kwargs for a spec — the single translation used by
     both ``create_blueprint_job`` and the suggestion path so they never drift."""
     return {
-        "prompt": spec.prompt,
-        "schedule": spec.schedule,
-        "name": name or f"blueprint:{spec.skill_name}",
-        "deliver": spec.deliver,
-        "skills": [spec.skill_name] if spec.skill_name else None,
-        "model": spec.model,
-        "provider": spec.provider,
-        "enabled_toolsets": spec.enabled_toolsets,
+        "prompt": spec.prompt, "schedule": spec.schedule, "name": name or f"blueprint:{spec.skill_name}",
+        "deliver": spec.deliver, "skills": [spec.skill_name] if spec.skill_name else None,
+        "model": spec.model, "provider": spec.provider, "enabled_toolsets": spec.enabled_toolsets,
         "no_agent": spec.no_agent,
     }
 
 
-def create_blueprint_job(
-    spec: BlueprintSpec,
-    *,
-    origin: Optional[Dict[str, Any]] = None,
-    name: Optional[str] = None,
-) -> Dict[str, Any]:
+def create_blueprint_job(spec: BlueprintSpec, *, origin: Optional[Dict[str, Any]] = None,
+                         name: Optional[str] = None) -> Dict[str, Any]:
     """Create the cron job for a spec (skill preloaded via ``skills=[name]``); returns the job dict."""
     from cron.scheduler import create_job_with_scheduler_registration
 
@@ -194,13 +147,10 @@ def register_blueprint_suggestion(spec: BlueprintSpec) -> Optional[Dict[str, Any
     except Exception:  # pragma: no cover - import guard
         return None
 
+    deliver = f", delivering to {spec.deliver}" if spec.deliver and spec.deliver != "origin" else ""
     return add_suggestion(
         title=f"Schedule '{spec.skill_name}'",
-        description=(
-            f"The '{spec.skill_name}' blueprint runs on schedule {spec.schedule}"
-            + (f", delivering to {spec.deliver}" if spec.deliver and spec.deliver != "origin" else "")
-            + "."
-        ),
+        description=f"The '{spec.skill_name}' blueprint runs on schedule {spec.schedule}{deliver}.",
         source="blueprint",
         job_spec=blueprint_to_job_spec(spec),
         dedup_key=f"blueprint:{spec.skill_name}:{spec.schedule}",
@@ -213,38 +163,24 @@ def export_blueprint(job: Dict[str, Any], body: str, *, blueprint_name: Optional
     ``body`` becomes the SKILL.md body; its first line is the description."""
     import yaml
 
-    name = blueprint_name or job.get("name") or "shared-blueprint"
     # Sanitize to a valid skill identifier.
-    name = "".join(c if (c.isalnum() or c in "-_") else "-" for c in str(name).lower())
-    name = name.strip("-_") or "shared-blueprint"
+    name = str(blueprint_name or job.get("name") or "shared-blueprint").lower()
+    name = "".join(c if (c.isalnum() or c in "-_") else "-" for c in name).strip("-_") or "shared-blueprint"
 
-    blueprint_block: Dict[str, Any] = {
-        "schedule": job.get("schedule_display") or _schedule_to_string(job.get("schedule")),
-    }
+    block: Dict[str, Any] = {"schedule": job.get("schedule_display") or _schedule_to_string(job.get("schedule"))}
     if job.get("deliver") and job["deliver"] != "origin":
-        blueprint_block["deliver"] = job["deliver"]
+        block["deliver"] = job["deliver"]
     if job.get("prompt"):
-        blueprint_block["prompt"] = job["prompt"]
+        block["prompt"] = job["prompt"]
     if job.get("no_agent"):
-        blueprint_block["no_agent"] = True
-    for key in ("model", "provider", "enabled_toolsets"):
-        if job.get(key):
-            blueprint_block[key] = job[key]
+        block["no_agent"] = True
+    block.update({k: job[k] for k in ("model", "provider", "enabled_toolsets") if job.get(k)})
 
     body = body.strip()
-    description = body.splitlines()[0][:200] if body else "Shared automation blueprint."
-
     frontmatter = {
-        "name": name,
-        "description": description,
-        "version": "1.0.0",
-        "license": "MIT",
-        "metadata": {
-            "hermes": {
-                "tags": ["blueprint", "automation"],
-                "blueprint": blueprint_block,
-            }
-        },
+        "name": name, "description": body.splitlines()[0][:200] if body else "Shared automation blueprint.",
+        "version": "1.0.0", "license": "MIT",
+        "metadata": {"hermes": {"tags": ["blueprint", "automation"], "blueprint": block}},
     }
     fm_yaml = yaml.safe_dump(frontmatter, sort_keys=False, allow_unicode=True).strip()
     body_text = body or f"# {name}\n\nShared automation blueprint."
@@ -260,14 +196,12 @@ def _schedule_to_string(schedule: Any) -> str:
         if kind == "cron" and schedule.get("expr"):
             return str(schedule["expr"])
         if kind == "interval":
-            # parse_schedule stores interval periods as "minutes"; tolerate a
-            # legacy/foreign "seconds" form too.
+            # parse_schedule stores interval periods as "minutes"; tolerate a legacy/foreign "seconds" form too.
             if schedule.get("minutes"):
                 mins = int(schedule["minutes"])
                 return f"every {mins // 60}h" if mins % 60 == 0 else f"every {mins}m"
             if schedule.get("seconds"):
                 secs = int(schedule["seconds"])
-                if secs % 3600 == 0:
-                    return f"every {secs // 3600}h"
-                return f"every {secs // 60}m" if secs % 60 == 0 else f"every {secs}s"
+                return (f"every {secs // 3600}h" if secs % 3600 == 0
+                        else f"every {secs // 60}m" if secs % 60 == 0 else f"every {secs}s")
     return "0 9 * * *"  # safe daily fallback
diff --git a/tools/credential_files.py b/tools/credential_files.py
index f0fe043757..550274d123 100644
--- a/tools/credential_files.py
+++ b/tools/credential_files.py
@@ -1,10 +1,8 @@
-"""File passthrough registry for remote terminal backends.
+"""File passthrough registry for remote terminal backends (Docker, Modal, SSH).
 
-Remote backends (Docker, Modal, SSH) create sandboxes with no host files.
-This module tells them which credential files (skill ``required_credential_files``
-+ ``terminal.credential_files`` config), skill directories, and host-side cache
-directories (documents, images, audio, screenshots, uploads) to mount or sync
-in, at sandbox creation and before each command (resync on Modal).
+Sandboxes start with no host files; this module tells them which credential files
+(skill ``required_credential_files`` + ``terminal.credential_files`` config), skill
+dirs, and host cache dirs to mount or sync in, at creation and before each command.
 """
 
 from __future__ import annotations
@@ -14,7 +12,7 @@ import os
 import posixpath
 from contextvars import ContextVar
 from pathlib import Path
-from typing import Dict, Iterator, List, Optional, Tuple
+from typing import Callable, Dict, Iterator, List, Optional, Tuple
 
 from hermes_cli.config import cfg_get
 from hermes_constants import get_hermes_dir, get_hermes_home
@@ -31,91 +29,67 @@ logger = logging.getLogger(__name__)
 # Session-scoped registry; ContextVar prevents cross-session bleed in the gateway.
 _registered_files_var: ContextVar[Dict[str, str]] = ContextVar("_registered_files")
 
-# Cache for config-based file list (loaded once per process).
+# Cache for config-based file list (loaded once per process; tests reset it).
 _config_files: List[Dict[str, str]] | None = None
+# Reused across calls so sanitized skill copies don't accumulate.
+_safe_skills_tempdir: Path | None = None
 
 
 def _get_registered() -> Dict[str, str]:
-    try:
-        return _registered_files_var.get()
-    except LookupError:
-        val: Dict[str, str] = {}
-        _registered_files_var.set(val)
-        return val
+    val = _registered_files_var.get(None)
+    if val is None:
+        _registered_files_var.set(val := {})
+    return val
 
 
 def _mount(host_path: Path | str, container_path: str) -> Dict[str, str]:
     return {"host_path": str(host_path), "container_path": container_path}
 
 
-def _contained_host_path(
-    rel: str, hermes_home: Path, abs_msg: str, traversal_msg: str
-) -> Optional[Path]:
+def _contained_host_path(rel: str, hermes_home: Path, abs_msg: str, traversal_msg: str) -> Optional[Path]:
     """Resolve *rel* under HERMES_HOME, refusing absolute paths and escapes."""
     if os.path.isabs(rel):
         logger.warning(abs_msg, rel)
         return None
     host_path = hermes_home / rel
-    # Resolve symlinks and ``..`` before the containment check.
-    from tools.path_security import validate_within_dir
+    from tools.path_security import validate_within_dir  # resolves symlinks and ``..`` before checking
 
-    containment_error = validate_within_dir(host_path, hermes_home)
-    if containment_error:
+    if containment_error := validate_within_dir(host_path, hermes_home):
         logger.warning(traversal_msg, rel, containment_error)
         return None
     return host_path.resolve()
 
 
-def register_credential_file(
-    relative_path: str,
-    container_base: str = "/root/.hermes",
-) -> bool:
-    """Register a HERMES_HOME-relative credential file for mounting.
+def register_credential_file(relative_path: str, container_base: str = "/root/.hermes") -> bool:
+    """Register a HERMES_HOME-relative credential file for mounting; True if it exists and was registered.
 
-    Returns True if the file exists on the host and was registered. Rejects
-    absolute paths and traversal out of HERMES_HOME. Containment alone is not
-    enough because HERMES_HOME holds the MASTER stores (``.env``, ``auth.json``,
-    ``mcp-tokens/``): those are refused via the canonical read deny-list
-    (``agent.file_safety.get_read_block_error``), so the mount surface cannot
-    hand a skill what the read surface denies it.
+    Rejects absolute paths and traversal out of HERMES_HOME. Containment alone is not
+    enough: HERMES_HOME holds the MASTER stores (``.env``, ``auth.json``, ``mcp-tokens/``),
+    which are refused via the canonical read deny-list so the mount surface cannot hand a
+    skill what the read surface denies. Fails CLOSED (logged) if the guard is unavailable or raises.
     """
     resolved = _contained_host_path(
-        relative_path,
-        get_hermes_home(),
+        relative_path, get_hermes_home(),
         "credential_files: rejected absolute path %r (must be relative to HERMES_HOME)",
-        "credential_files: rejected path traversal %r (%s)",
-    )
+        "credential_files: rejected path traversal %r (%s)")
     if resolved is None:
         return False
     if not resolved.is_file():
         logger.debug("credential_files: skipping %s (not found)", resolved)
         return False
-
-    # Master stores pass the containment check above, so the deny-list is the
-    # real gate. Fails CLOSED: if the guard can't be consulted, refuse rather
-    # than risk bind-mounting auth.json into a sandbox; the import sentinel +
-    # logger.exception keep guard failures debuggable, not silently swallowed.
     if get_read_block_error is None:
-        logger.error(
-            "credential_files: refusing %r — agent.file_safety could not be "
-            "imported, so the master-store deny-list cannot be consulted",
-            relative_path,
-        )
+        logger.error("credential_files: refusing %r — agent.file_safety could not be "
+                     "imported, so the master-store deny-list cannot be consulted", relative_path)
         return False
     try:
         denied = get_read_block_error(str(resolved))
     except Exception:
-        logger.exception(
-            "credential_files: refusing %r — read guard raised", relative_path
-        )
+        logger.exception("credential_files: refusing %r — read guard raised", relative_path)
         return False
     if denied:
-        logger.warning(
-            "credential_files: refused %r — it is a credential store the agent "
-            "is denied from reading; a skill may mount its own service token, "
-            "not the master key files",
-            relative_path,
-        )
+        logger.warning("credential_files: refused %r — it is a credential store the agent "
+                       "is denied from reading; a skill may mount its own service token, "
+                       "not the master key files", relative_path)
         return False
 
     container_path = f"{container_base.rstrip('/')}/{relative_path}"
@@ -124,19 +98,15 @@ def register_credential_file(
     return True
 
 
-def register_credential_files(
-    entries: list,
-    container_base: str = "/root/.hermes",
-) -> List[str]:
+def register_credential_files(entries: list, container_base: str = "/root/.hermes") -> List[str]:
     """Register skill-frontmatter entries (str or dict with ``path``); return missing paths."""
     missing = []
     for entry in entries:
-        if isinstance(entry, str):
-            rel_path = entry.strip()
-        elif isinstance(entry, dict):
-            rel_path = (entry.get("path") or entry.get("name") or "").strip()
-        else:
+        if isinstance(entry, dict):
+            entry = entry.get("path") or entry.get("name") or ""
+        elif not isinstance(entry, str):
             continue
+        rel_path = entry.strip()
         if rel_path and not register_credential_file(rel_path, container_base):
             missing.append(rel_path)
     return missing
@@ -154,15 +124,13 @@ def _load_config_files() -> List[Dict[str, str]]:
         hermes_home = get_hermes_home()
         cred_files = cfg_get(read_raw_config(), "terminal", "credential_files")
         for item in cred_files if isinstance(cred_files, list) else []:
-            if not (isinstance(item, str) and item.strip()):
+            rel = item.strip() if isinstance(item, str) else ""
+            if not rel:
                 continue
-            rel = item.strip()
             resolved_path = _contained_host_path(
-                rel,
-                hermes_home,
+                rel, hermes_home,
                 "credential_files: rejected absolute config path %r",
-                "credential_files: rejected config path traversal %r (%s)",
-            )
+                "credential_files: rejected config path traversal %r (%s)")
             if resolved_path is not None and resolved_path.is_file():
                 result.append(_mount(resolved_path, f"/root/.hermes/{rel}"))
     except Exception as e:
@@ -173,19 +141,12 @@ def _load_config_files() -> List[Dict[str, str]]:
 
 
 def get_credential_file_mounts() -> List[Dict[str, str]]:
-    """Skill-registered + config credential files as ``host_path``/``container_path`` dicts."""
-    mounts: Dict[str, str] = {}
-
-    # Re-check existence (file may have been deleted since registration).
-    for container_path, host_path in _get_registered().items():
-        if Path(host_path).is_file():
-            mounts[container_path] = host_path
-
+    """Skill-registered + config credential files as ``host_path``/``container_path`` dicts (re-checked for existence)."""
+    mounts = {cp: hp for cp, hp in _get_registered().items() if Path(hp).is_file()}
     for entry in _load_config_files():
-        cp = entry["container_path"]
-        if cp not in mounts and Path(entry["host_path"]).is_file():
-            mounts[cp] = entry["host_path"]
-
+        cp, hp = entry["container_path"], entry["host_path"]
+        if cp not in mounts and Path(hp).is_file():
+            mounts[cp] = hp
     return [_mount(hp, cp) for cp, hp in mounts.items()]
 
 
@@ -194,10 +155,8 @@ def get_credential_file_mounts() -> List[Dict[str, str]]:
 def _skill_dir_roots(container_base: str) -> Iterator[Tuple[Path, str]]:
     """Yield ``(host_dir, container_root)`` for every existing skills directory.
 
-    Local skills mount at ``<base>/skills``, external dirs at
-    ``<base>/external_skills/<i>``, trusted project-local dirs at
-    ``<base>/project_skills/<i>`` (separate namespace so container paths stay
-    stable if external_dirs change).
+    Local skills mount at ``<base>/skills``, external at ``<base>/external_skills/<i>``, trusted
+    project-local at ``<base>/project_skills/<i>`` (own namespace so paths stay stable if external_dirs change).
     """
     base = container_base.rstrip("/")
     skills_dir = get_hermes_home() / "skills"
@@ -207,122 +166,67 @@ def _skill_dir_roots(container_base: str) -> Iterator[Tuple[Path, str]]:
         from agent.skill_utils import get_external_skills_dirs, get_project_skills_dirs
     except ImportError:
         return
-    for label, dirs in (("external_skills", get_external_skills_dirs()),
-                        ("project_skills", get_project_skills_dirs())):
-        for idx, d in enumerate(dirs):
-            if d.is_dir():
-                yield d, f"{base}/{label}/{idx}"
+    for label, dirs in (("external_skills", get_external_skills_dirs()), ("project_skills", get_project_skills_dirs())):
+        yield from ((d, f"{base}/{label}/{idx}") for idx, d in enumerate(dirs) if d.is_dir())
 
 
-def _iter_regular_files(host_dir: Path, container_root: str) -> Iterator[Dict[str, str]]:
-    """Per-file mount entries under *host_dir*, skipping symlinks."""
-    for item in host_dir.rglob("*"):
-        if item.is_symlink() or not item.is_file():
-            continue
-        yield _mount(item, f"{container_root}/{item.relative_to(host_dir)}")
+def _walk_skill_tree(root: Path) -> Iterator[Tuple[Path, List[Path]]]:
+    """Yield ``(dir, regular_non_symlink_files)`` for every directory a sandbox should receive.
+
+    Prunes ``EXCLUDED_SKILL_DIRS`` *before* descending so bookkeeping/dependency trees (``.hub``,
+    ``.archive``, ``.curator_backups``, ``node_modules``, ``.git``, ...) the remote agent never reads
+    are never even walked; sync thus agrees with discovery on what is skill content. Deliberately
+    not ``is_excluded_skill_path()``: that also prunes ``references/``, ``templates/``, ``assets/``,
+    ``scripts/`` — progressive-disclosure files and bundled scripts the sandbox does execute.
+    """
+    for dirpath, dirnames, filenames in os.walk(root):
+        dirnames[:] = sorted(d for d in dirnames if d not in EXCLUDED_SKILL_DIRS)
+        base = Path(dirpath)
+        yield base, [f for f in (base / n for n in filenames) if not f.is_symlink() and f.is_file()]
 
 
-def get_skills_directory_mount(
-    container_base: str = "/root/.hermes",
-) -> list[Dict[str, str]]:
+def get_skills_directory_mount(container_base: str = "/root/.hermes") -> list[Dict[str, str]]:
     """Directory mount entries for all skill dirs (local + external + project).
 
-    Bind mounts follow symlinks, so a dir containing any symlink is replaced by
-    a sanitized temp copy (regular files only); symlink-free dirs are returned
-    directly with zero overhead.
+    Bind mounts follow symlinks, so a dir containing any symlink is replaced by a sanitized
+    temp copy (regular files only); symlink-free dirs are returned directly, zero overhead.
     """
-    return [
-        _mount(_safe_skills_path(host_dir), container_path)
-        for host_dir, container_path in _skill_dir_roots(container_base)
-    ]
-
-
-_safe_skills_tempdir: Path | None = None
+    return [_mount(_safe_skills_path(d), cp) for d, cp in _skill_dir_roots(container_base)]
 
 
 def _safe_skills_path(skills_dir: Path) -> str:
-    """Return *skills_dir* if symlink-free, else a sanitized temp copy."""
+    """Return *skills_dir* if symlink-free, else a sanitized temp copy (same exclusions as sync)."""
     global _safe_skills_tempdir
 
     symlinks = [p for p in skills_dir.rglob("*") if p.is_symlink()]
     if not symlinks:
         return str(skills_dir)
-
     for link in symlinks:
-        logger.warning("credential_files: skipping symlink in skills dir: %s -> %s",
-                       link, os.readlink(link))
+        logger.warning("credential_files: skipping symlink in skills dir: %s -> %s", link, os.readlink(link))
 
     import atexit
     import shutil
     import tempfile
 
-    # Reuse the same temp dir across calls to avoid accumulation.
     if _safe_skills_tempdir and _safe_skills_tempdir.is_dir():
         shutil.rmtree(_safe_skills_tempdir, ignore_errors=True)
+    safe_dir = _safe_skills_tempdir = Path(tempfile.mkdtemp(prefix="hermes-skills-safe-"))
 
-    safe_dir = Path(tempfile.mkdtemp(prefix="hermes-skills-safe-"))
-    _safe_skills_tempdir = safe_dir
-
-    # Same exclusion rule as the per-file sync path (_iter_syncable_files):
-    # the sanitized copy is what gets mounted, so it must not carry the
-    # bookkeeping trees either. Prune before descending so a multi-GB
-    # .curator_backups is never even walked.
-    for dirpath, dirnames, filenames in os.walk(skills_dir):
-        dirnames[:] = sorted(d for d in dirnames if d not in EXCLUDED_SKILL_DIRS)
-        base = Path(dirpath)
+    for base, files in _walk_skill_tree(skills_dir):
         (safe_dir / base.relative_to(skills_dir)).mkdir(parents=True, exist_ok=True)
-        for name in filenames:
-            item = base / name
-            if item.is_symlink() or not item.is_file():
-                continue
+        for item in files:
             shutil.copy2(str(item), str(safe_dir / item.relative_to(skills_dir)))
 
-    def _cleanup():
-        if safe_dir.is_dir():
-            shutil.rmtree(safe_dir, ignore_errors=True)
-
-    atexit.register(_cleanup)
+    atexit.register(lambda: safe_dir.is_dir() and shutil.rmtree(safe_dir, ignore_errors=True))
     logger.info("credential_files: created symlink-safe skills copy at %s", safe_dir)
     return str(safe_dir)
 
 
-def _iter_syncable_files(root: Path):
-    """Yield ``(path, rel)`` for every regular, non-symlink file under *root*
-    that a sandbox should receive.
-
-    Prunes ``agent.skill_utils.EXCLUDED_SKILL_DIRS`` *before* descending, so
-    the walk never enters local bookkeeping and dependency trees (``.hub``
-    download cache, ``.archive``, ``.curator_backups``, ``node_modules``,
-    ``__pycache__``, ``.git``, ...) that the remote agent never reads — the
-    sync path agrees with discovery on what counts as skill content.
-
-    This deliberately does not use ``is_excluded_skill_path()``, which also
-    prunes ``references/``, ``templates/``, ``assets/`` and ``scripts/``.
-    Those hold progressive-disclosure support files and bundled scripts the
-    sandbox does execute, so they must keep syncing.
-    """
-    for dirpath, dirnames, filenames in os.walk(root):
-        dirnames[:] = sorted(d for d in dirnames if d not in EXCLUDED_SKILL_DIRS)
-        base = Path(dirpath)
-        for name in filenames:
-            item = base / name
-            if item.is_symlink() or not item.is_file():
-                continue
-            yield item, item.relative_to(root)
-
-
-def iter_skills_files(
-    container_base: str = "/root/.hermes",
-) -> List[Dict[str, str]]:
-    """Per-file entries for all skills files (for backends that upload individually).
-
-    Skips symlinks and anything under EXCLUDED_SKILL_DIRS (see _iter_syncable_files).
-    """
-    return [
-        _mount(item, f"{container_root}/{rel}")
-        for host_dir, container_root in _skill_dir_roots(container_base)
-        for item, rel in _iter_syncable_files(host_dir)
-    ]
+def iter_skills_files(container_base: str = "/root/.hermes") -> List[Dict[str, str]]:
+    """Per-file entries for all skills files (for backends that upload individually)."""
+    return [_mount(item, f"{container_root}/{item.relative_to(host_dir)}")
+            for host_dir, container_root in _skill_dir_roots(container_base)
+            for _base, files in _walk_skill_tree(host_dir) for item in files]
 
 
 # --- Cache directory mounts (documents, images, audio, videos, screenshots) ---
@@ -336,12 +240,9 @@ _CACHE_DIRS: list[tuple[str, str]] = [
     ("cache/screenshots", "browser_screenshots"),
     ("cache/web", "web_cache"),
     ("cache/delegation", "delegation_cache"),
-    # Oversized tool results (tools/tool_result_storage.py); host side is the
-    # single canonical location.
-    ("cache/spillover", "cache/spillover"),
-    # Flat top-level desktop staging dirs (tui_gateway attach RPCs), not under
-    # cache/; no legacy alias, so both slots match. Mounted so vision / file
-    # tools inside sandbox containers can reach uploads and dropped files.
+    ("cache/spillover", "cache/spillover"),  # oversized tool results; host side is canonical
+    # Flat top-level desktop staging dirs (tui_gateway attach RPCs; no legacy alias),
+    # mounted so vision/file tools in sandboxes reach uploads and dropped files.
     ("images", "images"),
     ("attachments", "attachments"),
 ]
@@ -355,11 +256,9 @@ def _cache_dir_roots(container_base: str, *, create_missing: bool) -> Iterator[T
         if not host_dir.is_dir():
             if not create_missing:
                 continue
-            # Docker snapshots this list at container CREATION, so a dir that
-            # appears later would dangle for the container's life: create it
-            # now; an empty bind mount costs nothing. get_hermes_dir already
-            # picked new-vs-legacy, so creating its answer can't shadow a
-            # populated legacy dir.
+            # Docker snapshots this list at container CREATION, so a dir appearing later
+            # would dangle for the container's life: create it now (empty bind mount is free).
+            # get_hermes_dir already picked new-vs-legacy, so this can't shadow a legacy dir.
             try:
                 host_dir.mkdir(parents=True, exist_ok=True)
             except OSError:
@@ -367,44 +266,30 @@ def _cache_dir_roots(container_base: str, *, create_missing: bool) -> Iterator[T
         yield host_dir, f"{base}/{new_subpath}"
 
 
-def get_cache_directory_mounts(
-    container_base: str = "/root/.hermes",
-) -> List[Dict[str, str]]:
+def get_cache_directory_mounts(container_base: str = "/root/.hermes") -> List[Dict[str, str]]:
     """Bind-mount entries for each cache directory (host layout via ``get_hermes_dir``)."""
     return [_mount(h, c) for h, c in _cache_dir_roots(container_base, create_missing=True)]
 
 
-def map_cache_path_to_container(
-    host_path: str,
-    container_base: str = "/root/.hermes",
-) -> Optional[str]:
-    """POSIX container path for a host path under an auto-mounted cache dir, else None."""
-    path = Path(host_path)
+def _remap_cache_path(path: str, container_base: str, src: str, dst: str, join: Callable[[str, Path], str]) -> Optional[str]:
+    """Translate *path* from the *src* side of a cache mount to its *dst* side; None if unmounted."""
     for mount in get_cache_directory_mounts(container_base=container_base):
-        try:
-            rel = path.relative_to(mount["host_path"])
-        except ValueError:
-            continue
-        return posixpath.join(mount["container_path"], rel.as_posix())
+        if Path(path).is_relative_to(mount[src]):
+            return join(mount[dst], Path(path).relative_to(mount[src]))
     return None
 
 
-def from_agent_visible_cache_path(
-    container_path: str,
-    container_base: str = "/root/.hermes",
-) -> str:
+def map_cache_path_to_container(host_path: str, container_base: str = "/root/.hermes") -> Optional[str]:
+    """POSIX container path for a host path under an auto-mounted cache dir, else None."""
+    return _remap_cache_path(host_path, container_base, "host_path", "container_path", lambda root, rel: posixpath.join(root, rel.as_posix()))
+
+
+def from_agent_visible_cache_path(container_path: str, container_base: str = "/root/.hermes") -> str:
     """Inverse of :func:`to_agent_visible_cache_path`; unchanged unless Docker + cache dir."""
     if os.environ.get("TERMINAL_ENV", "local") != "docker":
         return container_path
-
-    path = Path(container_path)
-    for mount in get_cache_directory_mounts(container_base=container_base):
-        try:
-            rel = path.relative_to(mount["container_path"])
-        except ValueError:
-            continue
-        return str(Path(mount["host_path"]) / rel)
-    return container_path
+    mapped = _remap_cache_path(container_path, container_base, "container_path", "host_path", lambda root, rel: str(Path(root) / rel))
+    return mapped if mapped is not None else container_path
 
 
 # Backends whose file-sync lands under the remote home: ``~/.hermes`` is
@@ -412,19 +297,13 @@ def from_agent_visible_cache_path(
 _HOME_RELATIVE_BACKENDS = frozenset({"ssh", "daytona", "vercel_sandbox"})
 
 
-def to_agent_visible_cache_path(
-    host_path: str,
-    container_base: str = "/root/.hermes",
-) -> str:
-    """Translate a host cache path to where the active backend sees it.
+def to_agent_visible_cache_path(host_path: str, container_base: str = "/root/.hermes") -> str:
+    """Translate a host cache path to where the active backend (TERMINAL_ENV) sees it.
 
-    Per-backend base (mirrors ``_agent_cache_base_for_env`` in
-    tools/image_generation_tool.py): docker/modal mount/sync at
-    ``/root/.hermes``; ssh/daytona/vercel_sandbox under ``~/.hermes``; plugin
-    backends declare ``cache_path_base`` (None = host paths remain correct);
-    local/singularity/unknown stay unchanged (Apptainer auto-binds the host
-    home, so translation would dangle). Backend comes from TERMINAL_ENV, as in
-    terminal_tool._get_environment_config.
+    Mirrors ``_agent_cache_base_for_env`` in tools/image_generation_tool.py: docker/modal mount at
+    ``/root/.hermes``; ssh/daytona/vercel_sandbox under ``~/.hermes``; plugin backends declare
+    ``cache_path_base`` (None = host paths stay correct); local/singularity/unknown unchanged
+    (Apptainer auto-binds the host home, so translation would dangle).
     """
     backend = (os.environ.get("TERMINAL_ENV") or "local").strip().lower()
     if backend in _HOME_RELATIVE_BACKENDS:
@@ -443,15 +322,11 @@ def to_agent_visible_cache_path(
     return mapped if mapped is not None else host_path
 
 
-def iter_cache_files(
-    container_base: str = "/root/.hermes",
-) -> List[Dict[str, str]]:
+def iter_cache_files(container_base: str = "/root/.hermes") -> List[Dict[str, str]]:
     """Per-file cache entries (Modal upload/resync); skips symlinks."""
-    return [
-        entry
-        for host_dir, container_root in _cache_dir_roots(container_base, create_missing=False)
-        for entry in _iter_regular_files(host_dir, container_root)
-    ]
+    return [_mount(item, f"{root}/{item.relative_to(host_dir)}")
+            for host_dir, root in _cache_dir_roots(container_base, create_missing=False)
+            for item in host_dir.rglob("*") if not item.is_symlink() and item.is_file()]
 
 
 def clear_credential_files() -> None: