1  """Shared SKILL.md preprocessing helpers."""
  2  
  3  import logging
  4  import re
  5  import subprocess
  6  from pathlib import Path
  7  
  8  logger = logging.getLogger(__name__)
  9  
 10  # Matches ${HERMES_SKILL_DIR} / ${HERMES_SESSION_ID} tokens in SKILL.md.
 11  # Tokens that don't resolve (e.g. ${HERMES_SESSION_ID} with no session) are
 12  # left as-is so the user can debug them.
 13  _SKILL_TEMPLATE_RE = re.compile(r"\$\{(HERMES_SKILL_DIR|HERMES_SESSION_ID)\}")
 14  
 15  # Matches inline shell snippets like:  !`date +%Y-%m-%d`
 16  # Non-greedy, single-line only -- no newlines inside the backticks.
 17  _INLINE_SHELL_RE = re.compile(r"!`([^`\n]+)`")
 18  
 19  # Cap inline-shell output so a runaway command can't blow out the context.
 20  _INLINE_SHELL_MAX_OUTPUT = 4000
 21  
 22  
 23  def load_skills_config() -> dict:
 24      """Load the ``skills`` section of config.yaml (best-effort)."""
 25      try:
 26          from hermes_cli.config import load_config
 27  
 28          cfg = load_config() or {}
 29          skills_cfg = cfg.get("skills")
 30          if isinstance(skills_cfg, dict):
 31              return skills_cfg
 32      except Exception:
 33          logger.debug("Could not read skills config", exc_info=True)
 34      return {}
 35  
 36  
 37  def substitute_template_vars(
 38      content: str,
 39      skill_dir: Path | None,
 40      session_id: str | None,
 41  ) -> str:
 42      """Replace ${HERMES_SKILL_DIR} / ${HERMES_SESSION_ID} in skill content.
 43  
 44      Only substitutes tokens for which a concrete value is available --
 45      unresolved tokens are left in place so the author can spot them.
 46      """
 47      if not content:
 48          return content
 49  
 50      skill_dir_str = str(skill_dir) if skill_dir else None
 51  
 52      def _replace(match: re.Match) -> str:
 53          token = match.group(1)
 54          if token == "HERMES_SKILL_DIR" and skill_dir_str:
 55              return skill_dir_str
 56          if token == "HERMES_SESSION_ID" and session_id:
 57              return str(session_id)
 58          return match.group(0)
 59  
 60      return _SKILL_TEMPLATE_RE.sub(_replace, content)
 61  
 62  
 63  def run_inline_shell(command: str, cwd: Path | None, timeout: int) -> str:
 64      """Execute a single inline-shell snippet and return its stdout (trimmed).
 65  
 66      Failures return a short ``[inline-shell error: ...]`` marker instead of
 67      raising, so one bad snippet can't wreck the whole skill message.
 68      """
 69      try:
 70          completed = subprocess.run(
 71              ["bash", "-c", command],
 72              cwd=str(cwd) if cwd else None,
 73              capture_output=True,
 74              text=True,
 75              timeout=max(1, int(timeout)),
 76              check=False,
 77          )
 78      except subprocess.TimeoutExpired:
 79          return f"[inline-shell timeout after {timeout}s: {command}]"
 80      except FileNotFoundError:
 81          return "[inline-shell error: bash not found]"
 82      except Exception as exc:
 83          return f"[inline-shell error: {exc}]"
 84  
 85      output = (completed.stdout or "").rstrip("\n")
 86      if not output and completed.stderr:
 87          output = completed.stderr.rstrip("\n")
 88      if len(output) > _INLINE_SHELL_MAX_OUTPUT:
 89          output = output[:_INLINE_SHELL_MAX_OUTPUT] + "...[truncated]"
 90      return output
 91  
 92  
 93  def expand_inline_shell(
 94      content: str,
 95      skill_dir: Path | None,
 96      timeout: int,
 97  ) -> str:
 98      """Replace every !`cmd` snippet in ``content`` with its stdout.
 99  
100      Runs each snippet with the skill directory as CWD so relative paths in
101      the snippet work the way the author expects.
102      """
103      if "!`" not in content:
104          return content
105  
106      def _replace(match: re.Match) -> str:
107          cmd = match.group(1).strip()
108          if not cmd:
109              return ""
110          return run_inline_shell(cmd, skill_dir, timeout)
111  
112      return _INLINE_SHELL_RE.sub(_replace, content)
113  
114  
115  def preprocess_skill_content(
116      content: str,
117      skill_dir: Path | None,
118      session_id: str | None = None,
119      skills_cfg: dict | None = None,
120  ) -> str:
121      """Apply configured SKILL.md template and inline-shell preprocessing."""
122      if not content:
123          return content
124  
125      cfg = skills_cfg if isinstance(skills_cfg, dict) else load_skills_config()
126      if cfg.get("template_vars", True):
127          content = substitute_template_vars(content, skill_dir, session_id)
128      if cfg.get("inline_shell", False):
129          timeout = int(cfg.get("inline_shell_timeout", 10) or 10)
130          content = expand_inline_shell(content, skill_dir, timeout)
131      return content