diff --git a/agent/skill_commands.py b/agent/skill_commands.py index 916e203a1..6b73e83b3 100644 --- a/agent/skill_commands.py +++ b/agent/skill_commands.py @@ -7,11 +7,15 @@ can invoke skills via /skill-name commands. import json import logging import re -import subprocess from pathlib import Path from typing import Any, Dict, Optional from hermes_constants import display_hermes_home +from agent.skill_preprocessing import ( + expand_inline_shell as _expand_inline_shell, + load_skills_config as _load_skills_config, + substitute_template_vars as _substitute_template_vars, +) logger = logging.getLogger(__name__) @@ -20,111 +24,6 @@ _skill_commands: Dict[str, Dict[str, Any]] = {} _SKILL_INVALID_CHARS = re.compile(r"[^a-z0-9-]") _SKILL_MULTI_HYPHEN = re.compile(r"-{2,}") -# Matches ${HERMES_SKILL_DIR} / ${HERMES_SESSION_ID} tokens in SKILL.md. -# Tokens that don't resolve (e.g. ${HERMES_SESSION_ID} with no session) are -# left as-is so the user can debug them. -_SKILL_TEMPLATE_RE = re.compile(r"\$\{(HERMES_SKILL_DIR|HERMES_SESSION_ID)\}") - -# Matches inline shell snippets like: !`date +%Y-%m-%d` -# Non-greedy, single-line only — no newlines inside the backticks. -_INLINE_SHELL_RE = re.compile(r"!`([^`\n]+)`") - -# Cap inline-shell output so a runaway command can't blow out the context. -_INLINE_SHELL_MAX_OUTPUT = 4000 - - -def _load_skills_config() -> dict: - """Load the ``skills`` section of config.yaml (best-effort).""" - try: - from hermes_cli.config import load_config - - cfg = load_config() or {} - skills_cfg = cfg.get("skills") - if isinstance(skills_cfg, dict): - return skills_cfg - except Exception: - logger.debug("Could not read skills config", exc_info=True) - return {} - - -def _substitute_template_vars( - content: str, - skill_dir: Path | None, - session_id: str | None, -) -> str: - """Replace ${HERMES_SKILL_DIR} / ${HERMES_SESSION_ID} in skill content. - - Only substitutes tokens for which a concrete value is available — - unresolved tokens are left in place so the author can spot them. - """ - if not content: - return content - - skill_dir_str = str(skill_dir) if skill_dir else None - - def _replace(match: re.Match) -> str: - token = match.group(1) - if token == "HERMES_SKILL_DIR" and skill_dir_str: - return skill_dir_str - if token == "HERMES_SESSION_ID" and session_id: - return str(session_id) - return match.group(0) - - return _SKILL_TEMPLATE_RE.sub(_replace, content) - - -def _run_inline_shell(command: str, cwd: Path | None, timeout: int) -> str: - """Execute a single inline-shell snippet and return its stdout (trimmed). - - Failures return a short ``[inline-shell error: ...]`` marker instead of - raising, so one bad snippet can't wreck the whole skill message. - """ - try: - completed = subprocess.run( - ["bash", "-c", command], - cwd=str(cwd) if cwd else None, - capture_output=True, - text=True, - timeout=max(1, int(timeout)), - check=False, - ) - except subprocess.TimeoutExpired: - return f"[inline-shell timeout after {timeout}s: {command}]" - except FileNotFoundError: - return f"[inline-shell error: bash not found]" - except Exception as exc: - return f"[inline-shell error: {exc}]" - - output = (completed.stdout or "").rstrip("\n") - if not output and completed.stderr: - output = completed.stderr.rstrip("\n") - if len(output) > _INLINE_SHELL_MAX_OUTPUT: - output = output[:_INLINE_SHELL_MAX_OUTPUT] + "…[truncated]" - return output - - -def _expand_inline_shell( - content: str, - skill_dir: Path | None, - timeout: int, -) -> str: - """Replace every !`cmd` snippet in ``content`` with its stdout. - - Runs each snippet with the skill directory as CWD so relative paths in - the snippet work the way the author expects. - """ - if "!`" not in content: - return content - - def _replace(match: re.Match) -> str: - cmd = match.group(1).strip() - if not cmd: - return "" - return _run_inline_shell(cmd, skill_dir, timeout) - - return _INLINE_SHELL_RE.sub(_replace, content) - - def _load_skill_payload(skill_identifier: str, task_id: str | None = None) -> tuple[dict[str, Any], Path | None, str] | None: """Load a skill by name/path and return (loaded_payload, skill_dir, display_name).""" raw_identifier = (skill_identifier or "").strip() @@ -143,7 +42,9 @@ def _load_skill_payload(skill_identifier: str, task_id: str | None = None) -> tu else: normalized = raw_identifier.lstrip("/") - loaded_skill = json.loads(skill_view(normalized, task_id=task_id)) + loaded_skill = json.loads( + skill_view(normalized, task_id=task_id, preprocess=False) + ) except Exception: return None diff --git a/agent/skill_preprocessing.py b/agent/skill_preprocessing.py new file mode 100644 index 000000000..b95d1ddda --- /dev/null +++ b/agent/skill_preprocessing.py @@ -0,0 +1,131 @@ +"""Shared SKILL.md preprocessing helpers.""" + +import logging +import re +import subprocess +from pathlib import Path + +logger = logging.getLogger(__name__) + +# Matches ${HERMES_SKILL_DIR} / ${HERMES_SESSION_ID} tokens in SKILL.md. +# Tokens that don't resolve (e.g. ${HERMES_SESSION_ID} with no session) are +# left as-is so the user can debug them. +_SKILL_TEMPLATE_RE = re.compile(r"\$\{(HERMES_SKILL_DIR|HERMES_SESSION_ID)\}") + +# Matches inline shell snippets like: !`date +%Y-%m-%d` +# Non-greedy, single-line only -- no newlines inside the backticks. +_INLINE_SHELL_RE = re.compile(r"!`([^`\n]+)`") + +# Cap inline-shell output so a runaway command can't blow out the context. +_INLINE_SHELL_MAX_OUTPUT = 4000 + + +def load_skills_config() -> dict: + """Load the ``skills`` section of config.yaml (best-effort).""" + try: + from hermes_cli.config import load_config + + cfg = load_config() or {} + skills_cfg = cfg.get("skills") + if isinstance(skills_cfg, dict): + return skills_cfg + except Exception: + logger.debug("Could not read skills config", exc_info=True) + return {} + + +def substitute_template_vars( + content: str, + skill_dir: Path | None, + session_id: str | None, +) -> str: + """Replace ${HERMES_SKILL_DIR} / ${HERMES_SESSION_ID} in skill content. + + Only substitutes tokens for which a concrete value is available -- + unresolved tokens are left in place so the author can spot them. + """ + if not content: + return content + + skill_dir_str = str(skill_dir) if skill_dir else None + + def _replace(match: re.Match) -> str: + token = match.group(1) + if token == "HERMES_SKILL_DIR" and skill_dir_str: + return skill_dir_str + if token == "HERMES_SESSION_ID" and session_id: + return str(session_id) + return match.group(0) + + return _SKILL_TEMPLATE_RE.sub(_replace, content) + + +def run_inline_shell(command: str, cwd: Path | None, timeout: int) -> str: + """Execute a single inline-shell snippet and return its stdout (trimmed). + + Failures return a short ``[inline-shell error: ...]`` marker instead of + raising, so one bad snippet can't wreck the whole skill message. + """ + try: + completed = subprocess.run( + ["bash", "-c", command], + cwd=str(cwd) if cwd else None, + capture_output=True, + text=True, + timeout=max(1, int(timeout)), + check=False, + ) + except subprocess.TimeoutExpired: + return f"[inline-shell timeout after {timeout}s: {command}]" + except FileNotFoundError: + return "[inline-shell error: bash not found]" + except Exception as exc: + return f"[inline-shell error: {exc}]" + + output = (completed.stdout or "").rstrip("\n") + if not output and completed.stderr: + output = completed.stderr.rstrip("\n") + if len(output) > _INLINE_SHELL_MAX_OUTPUT: + output = output[:_INLINE_SHELL_MAX_OUTPUT] + "...[truncated]" + return output + + +def expand_inline_shell( + content: str, + skill_dir: Path | None, + timeout: int, +) -> str: + """Replace every !`cmd` snippet in ``content`` with its stdout. + + Runs each snippet with the skill directory as CWD so relative paths in + the snippet work the way the author expects. + """ + if "!`" not in content: + return content + + def _replace(match: re.Match) -> str: + cmd = match.group(1).strip() + if not cmd: + return "" + return run_inline_shell(cmd, skill_dir, timeout) + + return _INLINE_SHELL_RE.sub(_replace, content) + + +def preprocess_skill_content( + content: str, + skill_dir: Path | None, + session_id: str | None = None, + skills_cfg: dict | None = None, +) -> str: + """Apply configured SKILL.md template and inline-shell preprocessing.""" + if not 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) + return content diff --git a/tests/tools/test_skills_tool.py b/tests/tools/test_skills_tool.py index 3cdfa98a9..172e71209 100644 --- a/tests/tools/test_skills_tool.py +++ b/tests/tools/test_skills_tool.py @@ -347,6 +347,71 @@ class TestSkillView: assert result["name"] == "my-skill" assert "Step 1" in result["content"] + def test_skill_view_applies_template_vars(self, tmp_path): + with ( + patch("tools.skills_tool.SKILLS_DIR", tmp_path), + patch( + "agent.skill_preprocessing.load_skills_config", + return_value={"template_vars": True, "inline_shell": False}, + ), + ): + skill_dir = _make_skill( + tmp_path, + "templated", + body="Run ${HERMES_SKILL_DIR}/scripts/do.sh in ${HERMES_SESSION_ID}", + ) + raw = skill_view("templated", task_id="session-123") + + result = json.loads(raw) + assert result["success"] is True + assert f"Run {skill_dir}/scripts/do.sh in session-123" in result["content"] + assert "${HERMES_SKILL_DIR}" in result["raw_content"] + + def test_skill_view_applies_inline_shell_when_enabled(self, tmp_path): + with ( + patch("tools.skills_tool.SKILLS_DIR", tmp_path), + patch( + "agent.skill_preprocessing.load_skills_config", + return_value={ + "template_vars": True, + "inline_shell": True, + "inline_shell_timeout": 5, + }, + ), + ): + _make_skill( + tmp_path, + "dynamic", + body="Current date: !`printf 2026-04-24`", + ) + raw = skill_view("dynamic") + + result = json.loads(raw) + assert result["success"] is True + assert "Current date: 2026-04-24" in result["content"] + assert "!`printf 2026-04-24`" not in result["content"] + assert "!`printf 2026-04-24`" in result["raw_content"] + + def test_skill_view_leaves_inline_shell_literal_when_disabled(self, tmp_path): + with ( + patch("tools.skills_tool.SKILLS_DIR", tmp_path), + patch( + "agent.skill_preprocessing.load_skills_config", + return_value={"template_vars": True, "inline_shell": False}, + ), + ): + _make_skill( + tmp_path, + "static", + body="Current date: !`printf SHOULD_NOT_RUN`", + ) + raw = skill_view("static") + + result = json.loads(raw) + assert result["success"] is True + assert "Current date: !`printf SHOULD_NOT_RUN`" in result["content"] + assert "Current date: SHOULD_NOT_RUN" not in result["content"] + def test_view_nonexistent_skill(self, tmp_path): with patch("tools.skills_tool.SKILLS_DIR", tmp_path): _make_skill(tmp_path, "other-skill") diff --git a/tools/skills_tool.py b/tools/skills_tool.py index 8bf92ef08..447e70b33 100644 --- a/tools/skills_tool.py +++ b/tools/skills_tool.py @@ -743,6 +743,9 @@ def _serve_plugin_skill( skill_md: Path, namespace: str, bare: str, + *, + preprocess: bool = True, + session_id: str | None = None, ) -> str: """Read a plugin-provided skill, apply guards, return JSON.""" from hermes_cli.plugins import _get_disabled_plugins, get_plugin_manager @@ -812,11 +815,27 @@ def _serve_plugin_skill( except Exception: banner = "" + rendered_content = content + if preprocess: + try: + from agent.skill_preprocessing import preprocess_skill_content + + rendered_content = preprocess_skill_content( + content, + skill_md.parent, + session_id=session_id, + ) + except Exception: + logger.debug( + "Could not preprocess plugin skill %s:%s", namespace, bare, exc_info=True + ) + return json.dumps( { "success": True, "name": f"{namespace}:{bare}", - "content": f"{banner}{content}" if banner else content, + "content": f"{banner}{rendered_content}" if banner else rendered_content, + "raw_content": content, "description": description, "linked_files": None, "readiness_status": SkillReadinessStatus.AVAILABLE.value, @@ -825,7 +844,12 @@ def _serve_plugin_skill( ) -def skill_view(name: str, file_path: str = None, task_id: str = None) -> str: +def skill_view( + name: str, + file_path: str = None, + task_id: str = None, + preprocess: bool = True, +) -> str: """ View the content of a skill or a specific file within a skill directory. @@ -834,6 +858,9 @@ def skill_view(name: str, file_path: str = None, task_id: str = None) -> str: Qualified names like "plugin:skill" resolve to plugin-provided skills. file_path: Optional path to a specific file within the skill (e.g., "references/api.md") task_id: Optional task identifier used to probe the active backend + preprocess: Apply configured SKILL.md template and inline shell rendering + to main skill content. Internal slash/preload callers disable this + because they render the skill message themselves. Returns: JSON string with skill content or error message @@ -879,7 +906,13 @@ def skill_view(name: str, file_path: str = None, task_id: str = None) -> str: }, ensure_ascii=False, ) - return _serve_plugin_skill(plugin_skill_md, namespace, bare) + return _serve_plugin_skill( + plugin_skill_md, + namespace, + bare, + preprocess=preprocess, + session_id=task_id, + ) # Plugin exists but this specific skill is missing? available = pm.list_plugin_skills(namespace) @@ -1280,13 +1313,29 @@ def skill_view(name: str, file_path: str = None, task_id: str = None) -> str: exc_info=True, ) + rendered_content = content + if preprocess: + try: + from agent.skill_preprocessing import preprocess_skill_content + + rendered_content = preprocess_skill_content( + content, + skill_dir, + session_id=task_id, + ) + except Exception: + logger.debug( + "Could not preprocess skill content for %s", skill_name, exc_info=True + ) + result = { "success": True, "name": skill_name, "description": frontmatter.get("description", ""), "tags": tags, "related_skills": related_skills, - "content": content, + "content": rendered_content, + "raw_content": content, "path": rel_path, "skill_dir": str(skill_dir) if skill_dir else None, "linked_files": linked_files if linked_files else None,