fix(skills): key skill-command cache per platform to prevent cross-platform leaks

agent/skill_commands.py

@@ -7,14 +7,18 @@
 
 import json
 import logging
+import os
 import re
 from datetime import datetime
 from pathlib import Path
 from typing import Any, Dict, Optional
 
 logger = logging.getLogger(__name__)
 
-_skill_commands: Dict[str, Dict[str, Any]] = {}
+# Keyed by resolved platform (str) or None for the global/fallback view.
+# Using a per-platform cache prevents a long-lived multi-platform process from
+# serving one platform's disabled-skill set to another (fixes #14536).
+_skill_commands: Dict[Optional[str], Dict[str, Dict[str, Any]]] = {}
 _PLAN_SLUG_RE = re.compile(r"[^a-z0-9]+")
 # Patterns for sanitizing skill names into clean hyphen-separated slugs.
 _SKILL_INVALID_CHARS = re.compile(r"[^a-z0-9-]")
@@ -197,18 +201,44 @@ def _build_skill_message(
     return "\n".join(parts)
 
 
-def scan_skill_commands() -> Dict[str, Dict[str, Any]]:
+def _resolve_platform(platform: Optional[str] = None) -> Optional[str]:
+    """Resolve the active platform for skill-command cache keying.
+
+    Priority: explicit arg → HERMES_SESSION_PLATFORM context var →
+    HERMES_PLATFORM env var → None (global/fallback view).
+    """
+    if platform:
+        return platform
+    try:
+        from gateway.session_context import get_session_env
+        session_platform = get_session_env("HERMES_SESSION_PLATFORM", "")
+        if session_platform:
+            return session_platform
+    except Exception:
+        pass
+    return os.getenv("HERMES_PLATFORM") or None
+
+
+def scan_skill_commands(platform: Optional[str] = None) -> Dict[str, Dict[str, Any]]:
     """Scan ~/.hermes/skills/ and return a mapping of /command -> skill info.
 
+    Args:
+        platform: Explicit platform name (e.g. ``"telegram"``).  When
+            *None*, resolved from ``HERMES_SESSION_PLATFORM`` context var
+            or the ``HERMES_PLATFORM`` environment variable.  The result
+            is cached per resolved platform so different platforms in the
+            same process each see their own disabled-skill view.
+
     Returns:
         Dict mapping "/skill-name" to {name, description, skill_md_path, skill_dir}.
     """
     global _skill_commands
-    _skill_commands = {}
+    resolved_platform = _resolve_platform(platform)
+    _skill_commands[resolved_platform] = {}
     try:
         from tools.skills_tool import SKILLS_DIR, _parse_frontmatter, skill_matches_platform, _get_disabled_skill_names
         from agent.skill_utils import get_external_skills_dirs
-        disabled = _get_disabled_skill_names()
+        disabled = _get_disabled_skill_names(resolved_platform)
         seen_names: set = set()
 
         # Scan local dir first, then external dirs
@@ -249,7 +279,7 @@ def scan_skill_commands() -> Dict[str, Dict[str, Any]]:
                     cmd_name = _SKILL_MULTI_HYPHEN.sub('-', cmd_name).strip('-')
                     if not cmd_name:
                         continue
-                    _skill_commands[f"/{cmd_name}"] = {
+                    _skill_commands[resolved_platform][f"/{cmd_name}"] = {
                         "name": name,
                         "description": description or f"Invoke the {name} skill",
                         "skill_md_path": str(skill_md),
@@ -259,14 +289,20 @@ def scan_skill_commands() -> Dict[str, Dict[str, Any]]:
                     continue
     except Exception:
         pass
-    return _skill_commands
+    return _skill_commands[resolved_platform]
 
 
-def get_skill_commands() -> Dict[str, Dict[str, Any]]:
-    """Return the current skill commands mapping (scan first if empty)."""
-    if not _skill_commands:
-        scan_skill_commands()
-    return _skill_commands
+def get_skill_commands(platform: Optional[str] = None) -> Dict[str, Dict[str, Any]]:
+    """Return the skill commands mapping for the given platform.
+
+    Args:
+        platform: Explicit platform name.  When *None*, resolved from
+            session context or environment (same as ``scan_skill_commands``).
+    """
+    resolved_platform = _resolve_platform(platform)
+    if resolved_platform not in _skill_commands:
+        scan_skill_commands(resolved_platform)
+    return _skill_commands[resolved_platform]
 
 
 def resolve_skill_command_key(command: str) -> Optional[str]:

tests/agent/test_skill_commands.py

@@ -405,3 +405,98 @@ def test_plan_skill_message_can_include_runtime_save_path_note(self, tmp_path):
         assert "Add a /plan command" in msg
         assert ".hermes/plans/plan.md" in msg
         assert "Runtime note:" in msg
+
+
+class TestPlatformAwareCache:
+    """Regression tests for #14536 — per-platform skill command cache."""
+
+    def _make_skill_file(self, skills_dir, name):
+        d = Path(skills_dir) / name
+        d.mkdir(parents=True, exist_ok=True)
+        (d / "SKILL.md").write_text(
+            f"---\nname: {name}\ndescription: {name} skill\n---\nDo the thing.\n"
+        )
+
+    def _make_disabled_fn(self, rules: dict):
+        """Return a get_disabled_skill_names replacement honouring a simple rules dict.
+
+        rules keys are platform names (str) or None for the global list.
+        Each value is a list of disabled skill names.
+        """
+        import os
+        def _impl(platform=None):
+            from gateway.session_context import get_session_env
+            resolved = platform or os.getenv("HERMES_PLATFORM") or get_session_env("HERMES_SESSION_PLATFORM") or None
+            if resolved and resolved in rules:
+                return set(rules[resolved])
+            return set(rules.get(None, []))
+        return _impl
+
+    def test_different_platforms_get_independent_caches(self, tmp_path):
+        """Two platforms with different disabled lists must not share cache entries."""
+        from agent.skill_commands import get_skill_commands
+
+        skills_dir = tmp_path / "skills"
+        for name in ["alpha", "beta"]:
+            self._make_skill_file(skills_dir, name)
+
+        disabled_fn = self._make_disabled_fn({"telegram": ["beta"], "discord": ["alpha"]})
+
+        with patch("tools.skills_tool.SKILLS_DIR", skills_dir), \
+             patch("agent.skill_utils.get_external_skills_dirs", return_value=[]), \
+             patch("tools.skills_tool._get_disabled_skill_names", disabled_fn):
+
+            telegram_cmds = get_skill_commands(platform="telegram")
+            discord_cmds = get_skill_commands(platform="discord")
+
+        tg_names = {info["name"] for info in telegram_cmds.values()}
+        dc_names = {info["name"] for info in discord_cmds.values()}
+
+        assert "alpha" in tg_names, "telegram should have alpha"
+        assert "beta" not in tg_names, "telegram should not have beta (disabled)"
+        assert "beta" in dc_names, "discord should have beta"
+        assert "alpha" not in dc_names, "discord should not have alpha (disabled)"
+
+    def test_platform_none_uses_global_disabled_list(self, tmp_path):
+        """When platform is None, the global disabled list is respected."""
+        from agent.skill_commands import get_skill_commands
+
+        skills_dir = tmp_path / "skills"
+        for name in ["gamma", "delta"]:
+            self._make_skill_file(skills_dir, name)
+
+        disabled_fn = self._make_disabled_fn({None: ["delta"]})
+
+        with patch("tools.skills_tool.SKILLS_DIR", skills_dir), \
+             patch("agent.skill_utils.get_external_skills_dirs", return_value=[]), \
+             patch("tools.skills_tool._get_disabled_skill_names", disabled_fn), \
+             patch.dict(os.environ, {"HERMES_PLATFORM": ""}, clear=False):
+
+            cmds = get_skill_commands(platform=None)
+
+        names = {info["name"] for info in cmds.values()}
+        assert "gamma" in names
+        assert "delta" not in names
+
+    def test_rescan_updates_platform_cache_entry(self, tmp_path):
+        """Calling scan_skill_commands again replaces the cached entry for that platform."""
+        from agent.skill_commands import scan_skill_commands
+
+        skills_dir = tmp_path / "skills"
+        self._make_skill_file(skills_dir, "epsilon")
+
+        disabled_v1 = self._make_disabled_fn({"telegram": []})
+        disabled_v2 = self._make_disabled_fn({"telegram": ["epsilon"]})
+
+        with patch("tools.skills_tool.SKILLS_DIR", skills_dir), \
+             patch("agent.skill_utils.get_external_skills_dirs", return_value=[]), \
+             patch("tools.skills_tool._get_disabled_skill_names", disabled_v1):
+            cmds_v1 = scan_skill_commands(platform="telegram")
+
+        with patch("tools.skills_tool.SKILLS_DIR", skills_dir), \
+             patch("agent.skill_utils.get_external_skills_dirs", return_value=[]), \
+             patch("tools.skills_tool._get_disabled_skill_names", disabled_v2):
+            cmds_v2 = scan_skill_commands(platform="telegram")
+
+        assert "epsilon" in {i["name"] for i in cmds_v1.values()}
+        assert "epsilon" not in {i["name"] for i in cmds_v2.values()}