fix: sanitize Telegram help command mentions

This commit is contained in:
Kenny Wang 2026-05-03 15:45:56 -06:00 committed by Teknium
parent 6fda92aa7f
commit 222767e5e8
2 changed files with 109 additions and 2 deletions

View file

@ -49,6 +49,29 @@ from hermes_cli.config import cfg_get
_AGENT_CACHE_MAX_SIZE = 128 _AGENT_CACHE_MAX_SIZE = 128
_AGENT_CACHE_IDLE_TTL_SECS = 3600.0 # evict agents idle for >1h _AGENT_CACHE_IDLE_TTL_SECS = 3600.0 # evict agents idle for >1h
_PLATFORM_CONNECT_TIMEOUT_SECS_DEFAULT = 30.0 _PLATFORM_CONNECT_TIMEOUT_SECS_DEFAULT = 30.0
_TELEGRAM_COMMAND_MENTION_RE = re.compile(r"(?<![\w:/])/([A-Za-z0-9][A-Za-z0-9_-]*)")
def _telegramize_command_mentions(text: str, platform: Any) -> str:
"""Rewrite slash-command mentions to Telegram-valid command names.
Telegram Bot API command names allow only lowercase letters, digits, and
underscores. Keep other platform renderings unchanged, but normalize
Telegram help text so command mentions remain clickable/valid there.
"""
platform_value = getattr(platform, "value", platform)
if platform_value != "telegram":
return text
from hermes_cli.commands import _sanitize_telegram_name
def _replace(match: re.Match[str]) -> str:
sanitized = _sanitize_telegram_name(match.group(1))
return f"/{sanitized}" if sanitized else match.group(0)
return _TELEGRAM_COMMAND_MENTION_RE.sub(_replace, text)
# Only auto-continue interrupted gateway turns while the interruption is fresh. # Only auto-continue interrupted gateway turns while the interruption is fresh.
# Stale tool-tail/resume markers can otherwise revive an unrelated old task # Stale tool-tail/resume markers can otherwise revive an unrelated old task
# after a gateway restart when the user's next message starts new work. # after a gateway restart when the user's next message starts new work.
@ -7302,7 +7325,10 @@ class GatewayRunner:
lines.append(f"\n... and {len(sorted_cmds) - 10} more. Use `/commands` for the full paginated list.") lines.append(f"\n... and {len(sorted_cmds) - 10} more. Use `/commands` for the full paginated list.")
except Exception: except Exception:
pass pass
return "\n".join(lines) return _telegramize_command_mentions(
"\n".join(lines),
getattr(getattr(event, "source", None), "platform", None),
)
async def _handle_commands_command(self, event: MessageEvent) -> str: async def _handle_commands_command(self, event: MessageEvent) -> str:
"""Handle /commands [page] - paginated list of all commands and skills.""" """Handle /commands [page] - paginated list of all commands and skills."""
@ -7355,7 +7381,10 @@ class GatewayRunner:
lines.extend(["", " | ".join(nav_parts)]) lines.extend(["", " | ".join(nav_parts)])
if page != requested_page: if page != requested_page:
lines.append(f"_(Requested page {requested_page} was out of range, showing page {page}.)_") lines.append(f"_(Requested page {requested_page} was out of range, showing page {page}.)_")
return "\n".join(lines) return _telegramize_command_mentions(
"\n".join(lines),
getattr(getattr(event, "source", None), "platform", None),
)
async def _handle_model_command(self, event: MessageEvent) -> Optional[str]: async def _handle_model_command(self, event: MessageEvent) -> Optional[str]:
"""Handle /model command — switch model for this session. """Handle /model command — switch model for this session.

View file

@ -0,0 +1,78 @@
"""Gateway command help rendering tests."""
import pytest
from gateway.config import Platform
from gateway.platforms.base import MessageEvent
from gateway.session import SessionSource
def _make_event(text: str, platform: Platform) -> MessageEvent:
return MessageEvent(
text=text,
source=SessionSource(
platform=platform,
chat_id="chat-1",
user_id="user-1",
user_name="tester",
chat_type="dm",
),
)
def _make_runner():
from gateway.run import GatewayRunner
return object.__new__(GatewayRunner)
@pytest.mark.asyncio
async def test_help_sanitizes_slash_command_mentions_for_telegram(monkeypatch):
"""Telegram help output must not expose invalid uppercase/hyphenated slashes."""
monkeypatch.setattr(
"agent.skill_commands.get_skill_commands",
lambda: {
"/Linear": {"description": "Open Linear"},
"/Custom-Thing": {"description": "Run a custom thing"},
},
)
result = await _make_runner()._handle_help_command(
_make_event("/help", Platform.TELEGRAM)
)
assert "`/linear`" in result
assert "`/custom_thing`" in result
assert "`/Linear`" not in result
assert "`/Custom-Thing`" not in result
@pytest.mark.asyncio
async def test_commands_sanitizes_slash_command_mentions_for_telegram(monkeypatch):
"""Paginated Telegram /commands output uses Telegram-valid slash mentions."""
monkeypatch.setattr(
"agent.skill_commands.get_skill_commands",
lambda: {"/Linear": {"description": "Open Linear"}},
)
result = await _make_runner()._handle_commands_command(
_make_event("/commands 999", Platform.TELEGRAM)
)
assert "`/linear`" in result
assert "`/Linear`" not in result
@pytest.mark.asyncio
async def test_help_keeps_non_telegram_slash_command_mentions_unchanged(monkeypatch):
"""Only Telegram needs slash mentions rewritten to Telegram command names."""
monkeypatch.setattr(
"agent.skill_commands.get_skill_commands",
lambda: {"/Linear": {"description": "Open Linear"}},
)
result = await _make_runner()._handle_help_command(
_make_event("/help", Platform.DISCORD)
)
assert "`/Linear`" in result