mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-30 18:57:51 +00:00
dd39ec2694
A slash-skill invocation is persisted expanded — activation note plus the entire skill body. _history_to_messages is the single display projection every surface reads, so that payload rendered as a chat bubble anywhere a session was resumed. Project it here onto the invocation the user typed, and tag skill/bundle dispatches with the same string so the live send matches. Rewind and regenerate replay from what the transcript shows, so re-expand the invocation server-side before running the turn: the replayed prompt is identical to the original and no client ever holds the body.
812 lines
32 KiB
Python
812 lines
32 KiB
Python
"""Shared slash command helpers for skills.
|
|
|
|
Shared between CLI (cli.py) and gateway (gateway/run.py) so both surfaces
|
|
can invoke skills via /skill-name commands.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
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__)
|
|
|
|
_skill_commands: Dict[str, Dict[str, Any]] = {}
|
|
_skill_commands_platform: Optional[str] = None
|
|
# Patterns for sanitizing skill names into clean hyphen-separated slugs.
|
|
_SKILL_INVALID_CHARS = re.compile(r"[^a-z0-9-]")
|
|
_SKILL_MULTI_HYPHEN = re.compile(r"-{2,}")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Skill-scaffolding markers and the canonical extractor.
|
|
#
|
|
# When a user invokes a /skill (or /bundle), Hermes expands the turn into a
|
|
# model-facing message that embeds the full skill body plus scaffolding. That
|
|
# expanded text is what flows into the agent loop — and into memory providers
|
|
# via MemoryManager. Providers that store or embed the raw user turn (mem0,
|
|
# openviking, hindsight, retaindb, byterover, honcho, supermemory) would
|
|
# otherwise capture the entire skill body instead of what the user actually
|
|
# asked. ``extract_user_instruction_from_skill_message`` recovers just the
|
|
# user's instruction so memory stays clean.
|
|
#
|
|
# These markers MUST stay byte-identical to the builders below
|
|
# (``_build_skill_message`` here, ``build_bundle_invocation_message`` in
|
|
# agent/skill_bundles.py). They are co-located with the single-skill builder
|
|
# on purpose, and the bundle markers are asserted against the bundle builder in
|
|
# tests/openviking_plugin/test_openviking.py::test_skill_markers_match_hermes_scaffolding.
|
|
# ---------------------------------------------------------------------------
|
|
_SKILL_INVOCATION_PREFIX = "[IMPORTANT: The user has invoked the "
|
|
_SINGLE_SKILL_MARKER = "The full skill content is loaded below.]"
|
|
_SINGLE_SKILL_INSTRUCTION = (
|
|
"The user has provided the following instruction alongside the skill invocation: "
|
|
)
|
|
_RUNTIME_NOTE = "\n\n[Runtime note:"
|
|
_BUNDLE_MARKER = " skill bundle,"
|
|
_BUNDLE_USER_INSTRUCTION = "\nUser instruction: "
|
|
_BUNDLE_FIRST_SKILL_BLOCK = "\n\n[Loaded as part of the "
|
|
|
|
# The skill name sits in the first quoted span of the activation note, for both
|
|
# the single-skill and the bundle header ("work" / "/clean /work").
|
|
_SKILL_NAME_RE = re.compile(re.escape(_SKILL_INVOCATION_PREFIX) + r'"([^"]*)"')
|
|
|
|
# SQL LIKE pattern matching a skill-expanded turn, for listing queries that
|
|
# have to recognize scaffolding before the row reaches Python. The prefix
|
|
# contains no LIKE wildcards (`%`, `_`), so it needs no ESCAPE clause.
|
|
SKILL_SCAFFOLD_SQL_LIKE = _SKILL_INVOCATION_PREFIX + "%"
|
|
|
|
# Marks where a preview query joined the head and tail of a long scaffolded
|
|
# message. ``describe_skill_invocation`` may hand back a span that runs across
|
|
# the joint (a bundle instruction cut off by the head window); callers cut the
|
|
# description there rather than show the skill body on the far side.
|
|
SKILL_EXCERPT_JOINT = "\x1e"
|
|
|
|
|
|
def extract_user_instruction_from_skill_message(content: Any) -> Optional[str]:
|
|
"""Recover the user's instruction from a slash-skill-expanded turn.
|
|
|
|
Returns:
|
|
- The original string unchanged when it is NOT skill scaffolding
|
|
(a normal user message passes straight through).
|
|
- The extracted user instruction when the scaffolding carried one.
|
|
- ``None`` when the content is skill scaffolding with no user
|
|
instruction (i.e. a bare ``/skill`` invocation). Callers that feed
|
|
memory providers should skip the turn in that case — there is no
|
|
user content worth storing.
|
|
"""
|
|
if not isinstance(content, str):
|
|
return None
|
|
|
|
if not content.startswith(_SKILL_INVOCATION_PREFIX):
|
|
return content
|
|
|
|
if _BUNDLE_MARKER in content:
|
|
return _extract_bundle_user_instruction(content)
|
|
|
|
if _SINGLE_SKILL_MARKER in content:
|
|
return _extract_single_skill_user_instruction(content)
|
|
|
|
return None
|
|
|
|
|
|
def describe_skill_invocation(content: Any, separator: str = " — ") -> Optional[str]:
|
|
"""Render a slash-skill-expanded turn the way the user typed it.
|
|
|
|
The expanded message embeds the whole skill body, so any surface that
|
|
summarizes a user turn from its raw content — session titles, sidebar
|
|
previews, the ``/rewind`` picker — otherwise shows the skill's own prose
|
|
as if the user had written it. That is how a skill's opening line ends up
|
|
as a session title.
|
|
|
|
Returns ``"/work — fix the title leak"``, or ``"/work"`` for a bare
|
|
invocation, or ``None`` when *content* is not skill scaffolding (the
|
|
caller should then summarize it as an ordinary message).
|
|
|
|
*separator* joins the command and the instruction. Previews use the
|
|
default em dash; pass ``" "`` for the literal invocation the user typed,
|
|
which is what chat transcripts render.
|
|
"""
|
|
if not isinstance(content, str) or not content.startswith(_SKILL_INVOCATION_PREFIX):
|
|
return None
|
|
|
|
match = _SKILL_NAME_RE.match(content)
|
|
name = (match.group(1) if match else "").strip()
|
|
# Bundle headers already carry their typed "/a /b" keys; a single skill is
|
|
# a bare name.
|
|
label = name if name.startswith("/") else f"/{name}"
|
|
|
|
instruction = extract_user_instruction_from_skill_message(content)
|
|
if instruction and instruction is not content:
|
|
# An excerpted message (head + tail, joined by SKILL_EXCERPT_JOINT) can
|
|
# put the joint inside the matched span — keep only the side the
|
|
# instruction marker was found on.
|
|
instruction = instruction.split(SKILL_EXCERPT_JOINT)[0]
|
|
instruction = " ".join(instruction.split())
|
|
if instruction:
|
|
return f"{label}{separator}{instruction}" if name else instruction
|
|
|
|
return label if name else None
|
|
|
|
|
|
def _extract_single_skill_user_instruction(message: str) -> Optional[str]:
|
|
# Single-skill format appends the user instruction after the skill body, so
|
|
# the last occurrence is the user-provided one; the body may quote this text.
|
|
marker_idx = message.rfind(_SINGLE_SKILL_INSTRUCTION)
|
|
if marker_idx < 0:
|
|
return None
|
|
|
|
instruction = message[marker_idx + len(_SINGLE_SKILL_INSTRUCTION):]
|
|
runtime_idx = instruction.find(_RUNTIME_NOTE)
|
|
if runtime_idx >= 0:
|
|
instruction = instruction[:runtime_idx]
|
|
instruction = instruction.strip()
|
|
return instruction or None
|
|
|
|
|
|
def _extract_bundle_user_instruction(message: str) -> Optional[str]:
|
|
# Bundle format puts the user instruction before the loaded skills, so the
|
|
# first occurrence is the user-provided one.
|
|
marker_idx = message.find(_BUNDLE_USER_INSTRUCTION)
|
|
if marker_idx < 0:
|
|
return None
|
|
|
|
instruction = message[marker_idx + len(_BUNDLE_USER_INSTRUCTION):]
|
|
first_skill_idx = instruction.find(_BUNDLE_FIRST_SKILL_BLOCK)
|
|
if first_skill_idx >= 0:
|
|
instruction = instruction[:first_skill_idx]
|
|
instruction = instruction.strip()
|
|
return instruction or None
|
|
|
|
|
|
def _resolve_skill_commands_platform() -> Optional[str]:
|
|
"""Return the current platform scope used for disabled-skill filtering.
|
|
|
|
Used to detect when the active platform has shifted so
|
|
:func:`get_skill_commands` can drop a stale cache that was populated
|
|
for a different platform's ``skills.platform_disabled`` view (#14536).
|
|
|
|
Resolves from (in order) ``HERMES_PLATFORM`` env var and
|
|
``HERMES_SESSION_PLATFORM`` from the gateway session context. Returns
|
|
``None`` when no platform scope is active (e.g. classic CLI, RL
|
|
rollouts, standalone scripts).
|
|
"""
|
|
try:
|
|
from gateway.session_context import get_session_env
|
|
|
|
resolved_platform = (
|
|
os.getenv("HERMES_PLATFORM")
|
|
or get_session_env("HERMES_SESSION_PLATFORM")
|
|
)
|
|
except Exception:
|
|
resolved_platform = os.getenv("HERMES_PLATFORM")
|
|
return resolved_platform or None
|
|
|
|
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()
|
|
if not raw_identifier:
|
|
return None
|
|
|
|
try:
|
|
from tools.skills_tool import SKILLS_DIR, skill_view
|
|
from agent.skill_utils import normalize_skill_lookup_name
|
|
|
|
normalized = normalize_skill_lookup_name(raw_identifier)
|
|
|
|
loaded_skill = json.loads(
|
|
skill_view(normalized, task_id=task_id, preprocess=False)
|
|
)
|
|
except Exception:
|
|
return None
|
|
|
|
if not loaded_skill.get("success"):
|
|
return None
|
|
|
|
skill_name = str(loaded_skill.get("name") or normalized)
|
|
skill_path = str(loaded_skill.get("path") or "")
|
|
skill_dir = None
|
|
# Prefer the absolute skill_dir returned by skill_view() — this is
|
|
# correct for both local and external skills. Fall back to the old
|
|
# SKILLS_DIR-relative reconstruction only when skill_dir is absent
|
|
# (e.g. legacy skill_view responses).
|
|
abs_skill_dir = loaded_skill.get("skill_dir")
|
|
if abs_skill_dir:
|
|
skill_dir = Path(abs_skill_dir)
|
|
elif skill_path:
|
|
try:
|
|
skill_dir = SKILLS_DIR / Path(skill_path).parent
|
|
except Exception:
|
|
skill_dir = None
|
|
|
|
return loaded_skill, skill_dir, skill_name
|
|
|
|
|
|
def _inject_skill_config(loaded_skill: dict[str, Any], parts: list[str]) -> None:
|
|
"""Resolve and inject skill-declared config values into the message parts.
|
|
|
|
If the loaded skill's frontmatter declares ``metadata.hermes.config``
|
|
entries, their current values (from config.yaml or defaults) are appended
|
|
as a ``[Skill config: ...]`` block so the agent knows the configured values
|
|
without needing to read config.yaml itself.
|
|
"""
|
|
try:
|
|
from agent.skill_utils import (
|
|
extract_skill_config_vars,
|
|
parse_frontmatter,
|
|
resolve_skill_config_values,
|
|
)
|
|
|
|
# The loaded_skill dict contains the raw content which includes frontmatter
|
|
raw_content = str(loaded_skill.get("raw_content") or loaded_skill.get("content") or "")
|
|
if not raw_content:
|
|
return
|
|
|
|
frontmatter, _ = parse_frontmatter(raw_content)
|
|
config_vars = extract_skill_config_vars(frontmatter)
|
|
if not config_vars:
|
|
return
|
|
|
|
resolved = resolve_skill_config_values(config_vars)
|
|
if not resolved:
|
|
return
|
|
|
|
lines = ["", f"[Skill config (from {display_hermes_home()}/config.yaml):"]
|
|
for key, value in resolved.items():
|
|
display_val = str(value) if value else "(not set)"
|
|
lines.append(f" {key} = {display_val}")
|
|
lines.append("]")
|
|
parts.extend(lines)
|
|
except Exception:
|
|
pass # Non-critical — skill still loads without config injection
|
|
|
|
|
|
def _build_skill_message(
|
|
loaded_skill: dict[str, Any],
|
|
skill_dir: Path | None,
|
|
activation_note: str,
|
|
user_instruction: str = "",
|
|
runtime_note: str = "",
|
|
session_id: str | None = None,
|
|
) -> str:
|
|
"""Format a loaded skill into a user/system message payload."""
|
|
from tools.skills_tool import SKILLS_DIR
|
|
|
|
content = str(loaded_skill.get("content") or "")
|
|
|
|
# ── Template substitution and inline-shell expansion ──
|
|
# 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)
|
|
|
|
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(
|
|
"Resolve any relative paths in this skill (e.g. `scripts/foo.js`, "
|
|
"`templates/config.yaml`) against that directory, then run them "
|
|
"with the terminal tool using the absolute path."
|
|
)
|
|
|
|
# ── Inject resolved skill config values ──
|
|
_inject_skill_config(loaded_skill, parts)
|
|
|
|
if loaded_skill.get("setup_skipped"):
|
|
parts.extend(
|
|
[
|
|
"",
|
|
"[Skill setup note: Required environment setup was skipped. Continue loading the skill and explain any reduced functionality if it matters.]",
|
|
]
|
|
)
|
|
elif loaded_skill.get("gateway_setup_hint"):
|
|
parts.extend(
|
|
[
|
|
"",
|
|
f"[Skill setup note: {loaded_skill['gateway_setup_hint']}]",
|
|
]
|
|
)
|
|
elif loaded_skill.get("setup_needed") and loaded_skill.get("setup_note"):
|
|
parts.extend(
|
|
[
|
|
"",
|
|
f"[Skill setup note: {loaded_skill['setup_note']}]",
|
|
]
|
|
)
|
|
|
|
supporting = []
|
|
linked_files = loaded_skill.get("linked_files") or {}
|
|
for entries in linked_files.values():
|
|
if isinstance(entries, list):
|
|
supporting.extend(entries)
|
|
|
|
if not supporting and skill_dir:
|
|
for subdir in ("references", "templates", "scripts", "assets"):
|
|
subdir_path = skill_dir / subdir
|
|
if subdir_path.exists():
|
|
for f in sorted(subdir_path.rglob("*")):
|
|
if f.is_file() and not f.is_symlink():
|
|
rel = str(f.relative_to(skill_dir))
|
|
supporting.append(rel)
|
|
|
|
if supporting and skill_dir:
|
|
try:
|
|
skill_view_target = str(skill_dir.relative_to(SKILLS_DIR))
|
|
except ValueError:
|
|
# Skill is from an external dir — use the skill name instead
|
|
skill_view_target = skill_dir.name
|
|
parts.append("")
|
|
parts.append("[This skill has supporting files:]")
|
|
for sf in supporting:
|
|
parts.append(f"- {sf} -> {skill_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`)."
|
|
)
|
|
|
|
if user_instruction:
|
|
parts.append("")
|
|
parts.append(f"The user has provided the following instruction alongside the skill invocation: {user_instruction}")
|
|
|
|
if runtime_note:
|
|
parts.append("")
|
|
parts.append(f"[Runtime note: {runtime_note}]")
|
|
|
|
return "\n".join(parts)
|
|
|
|
|
|
def scan_skill_commands() -> Dict[str, Dict[str, Any]]:
|
|
"""Scan ~/.hermes/skills/ and return a mapping of /command -> skill info.
|
|
|
|
Returns:
|
|
Dict mapping "/skill-name" to {name, description, skill_md_path, skill_dir}.
|
|
"""
|
|
global _skill_commands, _skill_commands_platform
|
|
_skill_commands_platform = _resolve_skill_commands_platform()
|
|
_skill_commands = {}
|
|
try:
|
|
from tools.skills_tool import SKILLS_DIR, _parse_frontmatter, skill_matches_platform, skill_matches_environment, _get_disabled_skill_names
|
|
from agent.skill_utils import get_external_skills_dirs, iter_skill_index_files
|
|
from hermes_cli.commands import resolve_command
|
|
disabled = _get_disabled_skill_names()
|
|
seen_names: set = set()
|
|
|
|
# Scan local dir first, then external dirs
|
|
dirs_to_scan = []
|
|
if SKILLS_DIR.exists():
|
|
dirs_to_scan.append(SKILLS_DIR)
|
|
dirs_to_scan.extend(get_external_skills_dirs())
|
|
|
|
for scan_dir in dirs_to_scan:
|
|
for skill_md in iter_skill_index_files(scan_dir, "SKILL.md"):
|
|
if any(part in {'.git', '.github', '.hub', '.archive'} for part in skill_md.parts):
|
|
continue
|
|
try:
|
|
content = skill_md.read_text(encoding='utf-8')
|
|
frontmatter, body = _parse_frontmatter(content)
|
|
# Skip skills incompatible with the current OS platform
|
|
if not skill_matches_platform(frontmatter):
|
|
continue
|
|
# Skip skills not relevant to the current runtime env
|
|
# (kanban/docker/s6). Offer-time only; explicit load bypasses.
|
|
if not skill_matches_environment(frontmatter):
|
|
continue
|
|
name = frontmatter.get('name', skill_md.parent.name)
|
|
if name in seen_names:
|
|
continue
|
|
# Respect user's disabled skills config
|
|
if name in disabled:
|
|
continue
|
|
description = frontmatter.get('description', '')
|
|
if not description:
|
|
for line in body.strip().split('\n'):
|
|
line = line.strip()
|
|
if line and not line.startswith('#'):
|
|
description = line[:80]
|
|
break
|
|
seen_names.add(name)
|
|
# Normalize to hyphen-separated slug, stripping
|
|
# non-alnum chars (e.g. +, /) to avoid invalid
|
|
# Telegram command names downstream.
|
|
cmd_name = name.lower().replace(' ', '-').replace('_', '-')
|
|
cmd_name = _SKILL_INVALID_CHARS.sub('', cmd_name)
|
|
cmd_name = _SKILL_MULTI_HYPHEN.sub('-', cmd_name).strip('-')
|
|
if not cmd_name:
|
|
continue
|
|
# Skip if this skill's auto-generated /command collides
|
|
# with a core Hermes slash command (name or alias). The
|
|
# skill remains fully loadable via /skill <name>.
|
|
# Uses resolve_command() so aliases and case variants are
|
|
# covered without maintaining a separate cache.
|
|
if resolve_command(cmd_name) is not None:
|
|
logger.warning(
|
|
"Skill %r generates slash command '/%s' which "
|
|
"collides with a core Hermes command; skipping "
|
|
"auto-registration. Use '/skill %s' instead.",
|
|
name, cmd_name, name,
|
|
)
|
|
continue
|
|
# Dedup on the resolved slug, not just the raw name: two
|
|
# distinct frontmatter names can normalize to the same
|
|
# slug (e.g. "git_helper" vs "git-helper"). First-wins
|
|
# preserves local-before-external precedence.
|
|
cmd_key = f"/{cmd_name}"
|
|
if cmd_key in _skill_commands:
|
|
logger.warning(
|
|
"Skill %r maps to slash command %s already claimed "
|
|
"by %r; keeping the first and skipping this one.",
|
|
name, cmd_key, _skill_commands[cmd_key]["name"],
|
|
)
|
|
continue
|
|
_skill_commands[cmd_key] = {
|
|
"name": name,
|
|
"description": description or f"Invoke the {name} skill",
|
|
"skill_md_path": str(skill_md),
|
|
"skill_dir": str(skill_md.parent),
|
|
}
|
|
except Exception:
|
|
continue
|
|
except Exception:
|
|
pass
|
|
return _skill_commands
|
|
|
|
|
|
def get_skill_commands() -> Dict[str, Dict[str, Any]]:
|
|
"""Return the current skill commands mapping (scan first if empty).
|
|
|
|
Rescans when the active platform scope changes (e.g. a gateway
|
|
process serving Telegram and Discord concurrently) so each platform
|
|
sees its own ``skills.platform_disabled`` view (#14536).
|
|
"""
|
|
if (
|
|
not _skill_commands
|
|
or _skill_commands_platform != _resolve_skill_commands_platform()
|
|
):
|
|
scan_skill_commands()
|
|
return _skill_commands
|
|
|
|
|
|
def reload_skills() -> Dict[str, Any]:
|
|
"""Re-scan the skills directory and return a diff of what changed.
|
|
|
|
Rescans ``~/.hermes/skills/`` and any ``skills.external_dirs`` so the
|
|
slash-command map (``agent.skill_commands._skill_commands``) reflects
|
|
skills added or removed on disk.
|
|
|
|
This does NOT invalidate the skills system-prompt cache. Skills are
|
|
called by name via ``/skill-name``, ``skills_list``, or ``skill_view``
|
|
— they don't need to be in the system prompt for the model to use them.
|
|
Keeping the prompt cache intact preserves prefix caching across the
|
|
reload, so a user invoking ``/reload-skills`` pays no cache-reset cost.
|
|
|
|
Returns:
|
|
Dict with keys::
|
|
|
|
{
|
|
"added": [{"name": str, "description": str}, ...],
|
|
"removed": [{"name": str, "description": str}, ...],
|
|
"unchanged": [skill names present before and after],
|
|
"total": total skill count after rescan,
|
|
"commands": total /slash-skill count after rescan,
|
|
}
|
|
|
|
``description`` is the skill's full SKILL.md frontmatter
|
|
``description:`` field. Note: the system prompt skill index
|
|
truncates this to the first 57 chars; see ``extract_skill_description``.
|
|
"""
|
|
# Snapshot pre-reload state (name -> description) from the current
|
|
# slash-command cache. Using dicts lets the post-rescan diff carry
|
|
# descriptions for newly-visible or just-removed skills without a
|
|
# second disk walk.
|
|
def _snapshot(cmds: Dict[str, Dict[str, Any]]) -> Dict[str, str]:
|
|
out: Dict[str, str] = {}
|
|
for slash_key, info in cmds.items():
|
|
bare = slash_key.lstrip("/")
|
|
out[bare] = (info or {}).get("description") or ""
|
|
return out
|
|
|
|
before = _snapshot(_skill_commands)
|
|
|
|
# Rescan the skills dir. ``scan_skill_commands`` resets
|
|
# ``_skill_commands = {}`` internally and repopulates it.
|
|
new_commands = scan_skill_commands()
|
|
|
|
after = _snapshot(new_commands)
|
|
|
|
added_names = sorted(set(after) - set(before))
|
|
removed_names = sorted(set(before) - set(after))
|
|
unchanged = sorted(set(after) & set(before))
|
|
|
|
added = [{"name": n, "description": after[n]} for n in added_names]
|
|
# For removed skills, use the description we had cached pre-rescan
|
|
# (the skill file is gone so we can't re-read it).
|
|
removed = [{"name": n, "description": before[n]} for n in removed_names]
|
|
|
|
return {
|
|
"added": added,
|
|
"removed": removed,
|
|
"unchanged": unchanged,
|
|
"total": len(after),
|
|
"commands": len(new_commands),
|
|
}
|
|
|
|
|
|
def resolve_skill_command_key(command: str) -> Optional[str]:
|
|
"""Resolve a user-typed /command to its canonical skill_cmds key.
|
|
|
|
Skills are always stored with hyphens — ``scan_skill_commands`` normalizes
|
|
spaces and underscores to hyphens when building the key. Hyphens and
|
|
underscores are treated interchangeably in user input: this matches
|
|
``_check_unavailable_skill`` and accommodates Telegram bot-command names
|
|
(which disallow hyphens, so ``/claude-code`` is registered as
|
|
``/claude_code`` and comes back in the underscored form).
|
|
|
|
Returns the matching ``/slug`` key from ``get_skill_commands()`` or
|
|
``None`` if no match.
|
|
"""
|
|
if not command:
|
|
return None
|
|
cmd_key = f"/{command.replace('_', '-')}"
|
|
return cmd_key if cmd_key in get_skill_commands() else None
|
|
|
|
|
|
def build_skill_invocation_message(
|
|
cmd_key: str,
|
|
user_instruction: str = "",
|
|
task_id: str | None = None,
|
|
runtime_note: str = "",
|
|
) -> Optional[str]:
|
|
"""Build the user message content for a skill slash command invocation.
|
|
|
|
Args:
|
|
cmd_key: The command key including leading slash (e.g., "/gif-search").
|
|
user_instruction: Optional text the user typed after the command.
|
|
|
|
Returns:
|
|
The formatted message string, or None if the skill wasn't found.
|
|
"""
|
|
commands = get_skill_commands()
|
|
skill_info = commands.get(cmd_key)
|
|
if not skill_info:
|
|
return None
|
|
|
|
loaded = _load_skill_payload(skill_info["skill_dir"], task_id=task_id)
|
|
if not loaded:
|
|
return None
|
|
|
|
loaded_skill, skill_dir, skill_name = loaded
|
|
|
|
# Track active usage for Curator lifecycle management (#17782)
|
|
try:
|
|
from tools.skill_usage import bump_use
|
|
bump_use(skill_name)
|
|
except Exception:
|
|
pass # Non-critical — skill invocation proceeds regardless
|
|
|
|
activation_note = (
|
|
f'[IMPORTANT: The user has invoked the "{skill_name}" skill, indicating they want '
|
|
"you to follow its instructions. The full skill content is loaded below.]"
|
|
)
|
|
return _build_skill_message(
|
|
loaded_skill,
|
|
skill_dir,
|
|
activation_note,
|
|
user_instruction=user_instruction,
|
|
runtime_note=runtime_note,
|
|
session_id=task_id,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stacked slash-skill invocations — `/skill-a /skill-b do XYZ` loads every
|
|
# leading skill (up to _MAX_STACKED_SKILLS), not just the first.
|
|
#
|
|
# Inspired by Claude Code v2.1.199 (July 2, 2026): "Stacked slash-skill
|
|
# invocations like /skill-a /skill-b do XYZ now load all leading skills
|
|
# (up to 5), not just the first."
|
|
#
|
|
# The generated message deliberately reuses the BUNDLE scaffolding markers
|
|
# ("skill bundle," header + "[Loaded as part of the " block prefix) so
|
|
# extract_user_instruction_from_skill_message() recovers the user's
|
|
# instruction without any new marker plumbing — memory providers keep
|
|
# storing what the user actually asked, not N skill bodies.
|
|
# ---------------------------------------------------------------------------
|
|
_MAX_STACKED_SKILLS = 5
|
|
|
|
|
|
def split_stacked_skill_commands(rest: str) -> tuple[list[str], str]:
|
|
"""Consume additional leading ``/skill`` tokens from *rest*.
|
|
|
|
*rest* is the text that follows the FIRST matched skill command (the
|
|
caller has already resolved that one). Leading whitespace-delimited
|
|
tokens that start with ``/`` and resolve to installed skill commands are
|
|
consumed, up to ``_MAX_STACKED_SKILLS`` total leading skills (i.e. at
|
|
most ``_MAX_STACKED_SKILLS - 1`` extra keys here). Parsing stops at the
|
|
first token that is not a resolvable skill command — that token and
|
|
everything after it become the user instruction.
|
|
|
|
Returns:
|
|
``(extra_cmd_keys, remaining_instruction)`` where ``extra_cmd_keys``
|
|
are canonical ``/slug`` keys from :func:`get_skill_commands`.
|
|
"""
|
|
keys: list[str] = []
|
|
remaining = rest or ""
|
|
while len(keys) < _MAX_STACKED_SKILLS - 1:
|
|
stripped = remaining.lstrip()
|
|
if not stripped.startswith("/"):
|
|
break
|
|
parts = stripped.split(None, 1)
|
|
token = parts[0]
|
|
tail = parts[1] if len(parts) > 1 else ""
|
|
cmd_key = resolve_skill_command_key(token.lstrip("/"))
|
|
if cmd_key is None or cmd_key in keys:
|
|
break
|
|
keys.append(cmd_key)
|
|
remaining = tail
|
|
return keys, remaining.strip()
|
|
|
|
|
|
def build_stacked_skill_invocation_message(
|
|
cmd_keys: list[str],
|
|
user_instruction: str = "",
|
|
task_id: str | None = None,
|
|
) -> Optional[tuple[str, list[str], list[str]]]:
|
|
"""Build the user message for a stacked multi-skill slash invocation.
|
|
|
|
Args:
|
|
cmd_keys: Canonical ``/slug`` keys, in the order the user typed them.
|
|
user_instruction: Text remaining after the leading skill commands.
|
|
|
|
Returns:
|
|
``(message, loaded_skill_names, missing_skill_names)`` or ``None``
|
|
when no skill could be loaded at all.
|
|
"""
|
|
commands = get_skill_commands()
|
|
|
|
loaded_names: list[str] = []
|
|
missing: list[str] = []
|
|
skill_blocks: list[str] = []
|
|
seen: set[str] = set()
|
|
|
|
for cmd_key in cmd_keys:
|
|
if not cmd_key or cmd_key in seen:
|
|
continue
|
|
seen.add(cmd_key)
|
|
|
|
skill_info = commands.get(cmd_key)
|
|
if not skill_info:
|
|
missing.append(cmd_key.lstrip("/"))
|
|
continue
|
|
|
|
loaded = _load_skill_payload(skill_info["skill_dir"], task_id=task_id)
|
|
if not loaded:
|
|
missing.append(cmd_key.lstrip("/"))
|
|
continue
|
|
loaded_skill, skill_dir, skill_name = loaded
|
|
|
|
# Track active usage for Curator lifecycle management (#17782)
|
|
try:
|
|
from tools.skill_usage import bump_use
|
|
bump_use(skill_name)
|
|
except Exception:
|
|
pass # Non-critical
|
|
|
|
# NOTE: must start with "[Loaded as part of the " — that prefix is
|
|
# the bundle block marker the memory-scaffolding extractor cuts on.
|
|
activation_note = (
|
|
f'[Loaded as part of the stacked skill invocation "{skill_name}".]'
|
|
)
|
|
skill_blocks.append(
|
|
_build_skill_message(
|
|
loaded_skill,
|
|
skill_dir,
|
|
activation_note,
|
|
session_id=task_id,
|
|
)
|
|
)
|
|
loaded_names.append(skill_name)
|
|
|
|
if not skill_blocks:
|
|
return None
|
|
|
|
# Header — must contain " skill bundle," so the bundle-format extractor
|
|
# in extract_user_instruction_from_skill_message() applies unchanged.
|
|
typed = " ".join(k for k in cmd_keys if k)
|
|
header_lines = [
|
|
f'[IMPORTANT: The user has invoked the "{typed}" stacked skill bundle, '
|
|
f"loading {len(loaded_names)} skills together. Treat every skill below "
|
|
"as active guidance for this turn.]",
|
|
"",
|
|
f"Skills loaded: {', '.join(loaded_names)}",
|
|
]
|
|
if missing:
|
|
header_lines.append(f"Skills missing (skipped): {', '.join(missing)}")
|
|
if user_instruction:
|
|
header_lines.extend(["", f"User instruction: {user_instruction}"])
|
|
|
|
header = "\n".join(header_lines)
|
|
return ("\n\n".join([header, *skill_blocks]), loaded_names, missing)
|
|
|
|
|
|
def build_preloaded_skills_prompt(
|
|
skill_identifiers: list[str],
|
|
task_id: str | None = None,
|
|
) -> tuple[str, list[str], list[str]]:
|
|
"""Load one or more skills for session-wide CLI/TUI preloading.
|
|
|
|
Returns (prompt_text, loaded_skill_names, missing_identifiers).
|
|
|
|
Disabled skills are treated the same as missing ones: this loads via a
|
|
raw identifier straight into ``_load_skill_payload``, bypassing
|
|
``get_skill_commands()``'s scan-time disabled filter — mirrors the
|
|
bundle-invocation gate (#59156). Without this, ``hermes -s <skill>`` or
|
|
a deployment's ``HERMES_TUI_SKILLS`` env var could force-load a skill an
|
|
operator disabled via ``skills.disabled``/``skills.platform_disabled``.
|
|
"""
|
|
prompt_parts: list[str] = []
|
|
loaded_names: list[str] = []
|
|
missing: list[str] = []
|
|
|
|
try:
|
|
from agent.skill_utils import get_disabled_skill_names
|
|
disabled_names = get_disabled_skill_names()
|
|
except Exception:
|
|
disabled_names = set()
|
|
|
|
seen: set[str] = set()
|
|
for raw_identifier in skill_identifiers:
|
|
identifier = (raw_identifier or "").strip()
|
|
if not identifier or identifier in seen:
|
|
continue
|
|
seen.add(identifier)
|
|
|
|
loaded = _load_skill_payload(identifier, task_id=task_id)
|
|
if not loaded:
|
|
missing.append(identifier)
|
|
continue
|
|
|
|
loaded_skill, skill_dir, skill_name = loaded
|
|
|
|
if skill_name in disabled_names or identifier in disabled_names:
|
|
missing.append(identifier)
|
|
continue
|
|
|
|
# Track active usage for Curator lifecycle management (#17782)
|
|
try:
|
|
from tools.skill_usage import bump_use
|
|
bump_use(skill_name)
|
|
except Exception:
|
|
pass # Non-critical
|
|
|
|
activation_note = (
|
|
f'[IMPORTANT: The user launched this CLI session with the "{skill_name}" skill '
|
|
"preloaded. Treat its instructions as active guidance for the duration of this "
|
|
"session unless the user overrides them.]"
|
|
)
|
|
prompt_parts.append(
|
|
_build_skill_message(
|
|
loaded_skill,
|
|
skill_dir,
|
|
activation_note,
|
|
session_id=task_id,
|
|
)
|
|
)
|
|
loaded_names.append(skill_name)
|
|
|
|
return "\n\n".join(prompt_parts), loaded_names, missing
|