fix: follow-up fixes for salvaged PRs

- Fix GatewayApp → GatewayRunner import in api_server.py (PR #4976)
- Update launchd test assertions for new bootstrap/bootout/kickstart commands (PR #4892)
- Add nonlocal message declaration in run_sync() to fix UnboundLocalError (pre-existing scoping bug)

agent/skill_commands.py

@@ -217,6 +217,25 @@ def get_skill_commands() -> Dict[str, Dict[str, Any]]:
     return _skill_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 = "",

cron/scheduler.py

@@ -15,7 +15,6 @@ import logging
 import os
 import subprocess
 import sys
-import traceback
 
 # fcntl is Unix-only; on Windows use msvcrt for file locking
 try:
@@ -99,24 +98,26 @@ def _resolve_delivery_target(job: dict) -> Optional[dict]:
 
     if ":" in deliver:
         platform_name, rest = deliver.split(":", 1)
-        # Check for thread_id suffix (e.g. "telegram:-1003724596514:17")
-        if ":" in rest:
-            chat_id, thread_id = rest.split(":", 1)
+        platform_key = platform_name.lower()
+
+        from tools.send_message_tool import _parse_target_ref
+
+        parsed_chat_id, parsed_thread_id, is_explicit = _parse_target_ref(platform_key, rest)
+        if is_explicit:
+            chat_id, thread_id = parsed_chat_id, parsed_thread_id
         else:
             chat_id, thread_id = rest, None
 
         # Resolve human-friendly labels like "Alice (dm)" to real IDs.
-        # send_message(action="list") shows labels with display suffixes
-        # that aren't valid platform IDs (e.g. WhatsApp JIDs).
         try:
             from gateway.channel_directory import resolve_channel_name
-            target = chat_id
-            # Strip display suffix like " (dm)" or " (group)"
-            if target.endswith(")") and " (" in target:
-                target = target.rsplit(" (", 1)[0].strip()
-            resolved = resolve_channel_name(platform_name.lower(), target)
+            resolved = resolve_channel_name(platform_key, chat_id)
             if resolved:
-                chat_id = resolved
+                parsed_chat_id, parsed_thread_id, resolved_is_explicit = _parse_target_ref(platform_key, resolved)
+                if resolved_is_explicit:
+                    chat_id, thread_id = parsed_chat_id, parsed_thread_id
+                else:
+                    chat_id = resolved
         except Exception:
             pass
 
@@ -628,7 +629,7 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
         
     except Exception as e:
         error_msg = f"{type(e).__name__}: {str(e)}"
-        logger.error("Job '%s' failed: %s", job_name, error_msg)
+        logger.exception("Job '%s' failed: %s", job_name, error_msg)
         
         output = f"""# Cron Job: {job_name} (FAILED)
 
@@ -644,8 +645,6 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
 
 ```
 {error_msg}
-
-{traceback.format_exc()}
 ```
 """
         return False, output, "", error_msg

gateway/channel_directory.py

@@ -18,6 +18,20 @@ logger = logging.getLogger(__name__)
 DIRECTORY_PATH = get_hermes_home() / "channel_directory.json"
 
 
+def _normalize_channel_query(value: str) -> str:
+    return value.lstrip("#").strip().lower()
+
+
+def _channel_target_name(platform_name: str, channel: Dict[str, Any]) -> str:
+    """Return the human-facing target label shown to users for a channel entry."""
+    name = channel["name"]
+    if platform_name == "discord" and channel.get("guild"):
+        return f"#{name}"
+    if platform_name != "discord" and channel.get("type"):
+        return f"{name} ({channel['type']})"
+    return name
+
+
 def _session_entry_id(origin: Dict[str, Any]) -> Optional[str]:
     chat_id = origin.get("chat_id")
     if not chat_id:
@@ -188,23 +202,25 @@ def resolve_channel_name(platform_name: str, name: str) -> Optional[str]:
     if not channels:
         return None
 
-    query = name.lstrip("#").lower()
+    query = _normalize_channel_query(name)
 
-    # 1. Exact name match
+    # 1. Exact name match, including the display labels shown by send_message(action="list")
     for ch in channels:
-        if ch["name"].lower() == query:
+        if _normalize_channel_query(ch["name"]) == query:
+            return ch["id"]
+        if _normalize_channel_query(_channel_target_name(platform_name, ch)) == query:
             return ch["id"]
 
     # 2. Guild-qualified match for Discord ("GuildName/channel")
     if "/" in query:
         guild_part, ch_part = query.rsplit("/", 1)
         for ch in channels:
-            guild = ch.get("guild", "").lower()
-            if guild == guild_part and ch["name"].lower() == ch_part:
+            guild = ch.get("guild", "").strip().lower()
+            if guild == guild_part and _normalize_channel_query(ch["name"]) == ch_part:
                 return ch["id"]
 
     # 3. Partial prefix match (only if unambiguous)
-    matches = [ch for ch in channels if ch["name"].lower().startswith(query)]
+    matches = [ch for ch in channels if _normalize_channel_query(ch["name"]).startswith(query)]
     if len(matches) == 1:
         return matches[0]["id"]
 
@@ -239,17 +255,16 @@ def format_directory_for_display() -> str:
             for guild_name, guild_channels in sorted(guilds.items()):
                 lines.append(f"Discord ({guild_name}):")
                 for ch in sorted(guild_channels, key=lambda c: c["name"]):
-                    lines.append(f"  discord:#{ch['name']}")
+                    lines.append(f"  discord:{_channel_target_name(plat_name, ch)}")
             if dms:
                 lines.append("Discord (DMs):")
                 for ch in dms:
-                    lines.append(f"  discord:{ch['name']}")
+                    lines.append(f"  discord:{_channel_target_name(plat_name, ch)}")
             lines.append("")
         else:
             lines.append(f"{plat_name.title()}:")
             for ch in channels:
-                type_label = f" ({ch['type']})" if ch.get("type") else ""
-                lines.append(f"  {plat_name}:{ch['name']}{type_label}")
+                lines.append(f"  {plat_name}:{_channel_target_name(plat_name, ch)}")
             lines.append("")
 
     lines.append('Use these as the "target" parameter when sending.')

gateway/platforms/api_server.py

@@ -421,6 +421,11 @@ class APIServerAdapter(BasePlatformAdapter):
 
         max_iterations = int(os.getenv("HERMES_MAX_ITERATIONS", "90"))
 
+        # Load fallback provider chain so the API server platform has the
+        # same fallback behaviour as Telegram/Discord/Slack (fixes #4954).
+        from gateway.run import GatewayRunner
+        fallback_model = GatewayRunner._load_fallback_model()
+
         agent = AIAgent(
             model=model,
             **runtime_kwargs,
@@ -434,6 +439,7 @@ class APIServerAdapter(BasePlatformAdapter):
             stream_delta_callback=stream_delta_callback,
             tool_progress_callback=tool_progress_callback,
             session_db=self._ensure_session_db(),
+            fallback_model=fallback_model,
         )
         return agent
 

gateway/platforms/base.py

@@ -1068,6 +1068,28 @@ class BasePlatformAdapter(ABC):
                     logger.error("[%s] Approval dispatch failed: %s", self.name, e, exc_info=True)
                 return
 
+            # /status must also bypass the active-session guard so it always
+            # returns a system-generated response instead of being queued as
+            # user text and passed to the agent (#5046).
+            if cmd == "status":
+                logger.debug(
+                    "[%s] Status command bypassing active-session guard for %s",
+                    self.name, session_key,
+                )
+                try:
+                    _thread_meta = {"thread_id": event.source.thread_id} if event.source.thread_id else None
+                    response = await self._message_handler(event)
+                    if response:
+                        await self._send_with_retry(
+                            chat_id=event.source.chat_id,
+                            content=response,
+                            reply_to=event.message_id,
+                            metadata=_thread_meta,
+                        )
+                except Exception as e:
+                    logger.error("[%s] Status dispatch failed: %s", self.name, e, exc_info=True)
+                return
+
             # Special case: photo bursts/albums frequently arrive as multiple near-
             # simultaneous messages. Queue them without interrupting the active run,
             # then process them immediately after the current task finishes.

gateway/platforms/telegram.py

@@ -601,6 +601,12 @@ class TelegramAdapter(BasePlatformAdapter):
                 )
             else:
                 # ── Polling mode (default) ───────────────────────────
+                # Clear any stale webhook first so polling doesn't inherit a
+                # previous webhook registration and silently stop receiving updates.
+                delete_webhook = getattr(self._bot, "delete_webhook", None)
+                if callable(delete_webhook):
+                    await delete_webhook(drop_pending_updates=False)
+
                 loop = asyncio.get_running_loop()
 
                 def _polling_error_callback(error: Exception) -> None:
@@ -856,6 +862,21 @@ class TelegramAdapter(BasePlatformAdapter):
                             await asyncio.sleep(wait)
                         else:
                             raise
+                    except Exception as send_err:
+                        retry_after = getattr(send_err, "retry_after", None)
+                        if retry_after is not None or "retry after" in str(send_err).lower():
+                            if _send_attempt < 2:
+                                wait = float(retry_after) if retry_after is not None else 1.0
+                                logger.warning(
+                                    "[%s] Telegram flood control on send (attempt %d/3), retrying in %.1fs: %s",
+                                    self.name,
+                                    _send_attempt + 1,
+                                    wait,
+                                    send_err,
+                                )
+                                await asyncio.sleep(wait)
+                                continue
+                        raise
                 message_ids.append(str(msg.message_id))
             
             return SendResult(

gateway/run.py

@@ -1266,6 +1266,8 @@ class GatewayRunner:
         next message, so there's no blocking delay.
         """
         await asyncio.sleep(60)  # initial delay — let the gateway fully start
+        _flush_failures: dict[str, int] = {}  # session_id -> consecutive failure count
+        _MAX_FLUSH_RETRIES = 3
         while self._running:
             try:
                 self.session_store._ensure_loaded()
@@ -1298,8 +1300,25 @@ class GatewayRunner:
                             "Pre-reset memory flush completed for session %s",
                             entry.session_id,
                         )
+                        _flush_failures.pop(entry.session_id, None)
                     except Exception as e:
-                        logger.debug("Proactive memory flush failed for %s: %s", entry.session_id, e)
+                        failures = _flush_failures.get(entry.session_id, 0) + 1
+                        _flush_failures[entry.session_id] = failures
+                        if failures >= _MAX_FLUSH_RETRIES:
+                            logger.warning(
+                                "Proactive memory flush gave up after %d attempts for %s: %s. "
+                                "Marking as flushed to prevent infinite retry loop.",
+                                failures, entry.session_id, e,
+                            )
+                            with self.session_store._lock:
+                                entry.memory_flushed = True
+                                self.session_store._save()
+                            _flush_failures.pop(entry.session_id, None)
+                        else:
+                            logger.debug(
+                                "Proactive memory flush failed (%d/%d) for %s: %s",
+                                failures, _MAX_FLUSH_RETRIES, entry.session_id, e,
+                            )
             except Exception as e:
                 logger.debug("Session expiry watcher error: %s", e)
             # Sleep in small increments so we can stop quickly
@@ -2093,7 +2112,10 @@ class GatewayRunner:
         if command:
             try:
                 from hermes_cli.plugins import get_plugin_command_handler
-                plugin_handler = get_plugin_command_handler(command)
+                # Normalize underscores to hyphens so Telegram's underscored
+                # autocomplete form matches plugin commands registered with
+                # hyphens. See hermes_cli/commands.py:_build_telegram_menu.
+                plugin_handler = get_plugin_command_handler(command.replace("_", "-"))
                 if plugin_handler:
                     user_args = event.get_command_args().strip()
                     import asyncio as _aio
@@ -2104,13 +2126,20 @@ class GatewayRunner:
             except Exception as e:
                 logger.debug("Plugin command dispatch failed (non-fatal): %s", e)
 
-        # Skill slash commands: /skill-name loads the skill and sends to agent
+        # Skill slash commands: /skill-name loads the skill and sends to agent.
+        # resolve_skill_command_key() handles the Telegram underscore/hyphen
+        # round-trip so /claude_code from Telegram autocomplete still resolves
+        # to the claude-code skill.
         if command:
             try:
-                from agent.skill_commands import get_skill_commands, build_skill_invocation_message
+                from agent.skill_commands import (
+                    get_skill_commands,
+                    build_skill_invocation_message,
+                    resolve_skill_command_key,
+                )
                 skill_cmds = get_skill_commands()
-                cmd_key = f"/{command}"
-                if cmd_key in skill_cmds:
+                cmd_key = resolve_skill_command_key(command)
+                if cmd_key is not None:
                     # Check per-platform disabled status before executing.
                     # get_skill_commands() only applies the *global* disabled
                     # list at scan time; per-platform overrides need checking
@@ -2137,6 +2166,27 @@ class GatewayRunner:
                     _unavail_msg = _check_unavailable_skill(command)
                     if _unavail_msg:
                         return _unavail_msg
+                    # Genuinely unrecognized /command: not a built-in, not a
+                    # plugin, not a skill, not a known-inactive skill. Warn
+                    # the user instead of silently forwarding it to the LLM
+                    # as free text (which leads to silent-failure behavior
+                    # like the model inventing a delegate_task call).
+                    # Normalize to hyphenated form before checking known
+                    # built-ins (command may be an alias target set by the
+                    # quick-command block above, so _cmd_def can be stale).
+                    if command.replace("_", "-") not in GATEWAY_KNOWN_COMMANDS:
+                        logger.warning(
+                            "Unrecognized slash command /%s from %s — "
+                            "replying with unknown-command notice",
+                            command,
+                            source.platform.value if source.platform else "?",
+                        )
+                        return (
+                            f"Unknown command `/{command}`. "
+                            f"Type /commands to see what's available, "
+                            f"or resend without the leading slash to send "
+                            f"as a regular message."
+                        )
             except Exception as e:
                 logger.debug("Skill command check failed (non-fatal): %s", e)
         
@@ -6183,6 +6233,14 @@ class GatewayRunner:
                 logger.debug("status_callback error (%s): %s", event_type, _e)
 
         def run_sync():
+            # The conditional re-assignment of `message` further below
+            # (prepending model-switch notes) makes Python treat it as a
+            # local variable in the entire function.  `nonlocal` lets us
+            # read *and* reassign the outer `_run_agent` parameter without
+            # triggering an UnboundLocalError on the earlier read at
+            # `_resolve_turn_agent_config(message, …)`.
+            nonlocal message
+
             # Pass session_key to process registry via env var so background
             # processes can be mapped back to this gateway session
             os.environ["HERMES_SESSION_KEY"] = session_key or ""

hermes_cli/gateway.py

@@ -873,6 +873,11 @@ def get_launchd_label() -> str:
     return f"ai.hermes.gateway-{suffix}" if suffix else "ai.hermes.gateway"
 
 
+def _launchd_domain() -> str:
+    import os
+    return f"gui/{os.getuid()}"
+
+
 def generate_launchd_plist() -> str:
     python_path = get_python_path()
     working_dir = str(PROJECT_ROOT)
@@ -963,18 +968,19 @@ def launchd_plist_is_current() -> bool:
 def refresh_launchd_plist_if_needed() -> bool:
     """Rewrite the installed launchd plist when the generated definition has changed.
 
-    Unlike systemd, launchd picks up plist changes on the next ``launchctl stop``/
-    ``launchctl start`` cycle — no daemon-reload is needed.  We still unload/reload
-    to make launchd re-read the updated plist immediately.
+    Unlike systemd, launchd picks up plist changes on the next ``launchctl kill``/
+    ``launchctl kickstart`` cycle — no daemon-reload is needed. We still bootout/
+    bootstrap to make launchd re-read the updated plist immediately.
     """
     plist_path = get_launchd_plist_path()
     if not plist_path.exists() or launchd_plist_is_current():
         return False
 
     plist_path.write_text(generate_launchd_plist(), encoding="utf-8")
-    # Unload/reload so launchd picks up the new definition
-    subprocess.run(["launchctl", "unload", str(plist_path)], check=False)
-    subprocess.run(["launchctl", "load", str(plist_path)], check=False)
+    label = get_launchd_label()
+    # Bootout/bootstrap so launchd picks up the new definition
+    subprocess.run(["launchctl", "bootout", f"{_launchd_domain()}/{label}"], check=False)
+    subprocess.run(["launchctl", "bootstrap", _launchd_domain(), str(plist_path)], check=False)
     print("↻ Updated gateway launchd service definition to match the current Hermes install")
     return True
 
@@ -996,7 +1002,7 @@ def launchd_install(force: bool = False):
     print(f"Installing launchd service to: {plist_path}")
     plist_path.write_text(generate_launchd_plist())
     
-    subprocess.run(["launchctl", "load", str(plist_path)], check=True)
+    subprocess.run(["launchctl", "bootstrap", _launchd_domain(), str(plist_path)], check=True)
     
     print()
     print("✓ Service installed and loaded!")
@@ -1008,7 +1014,8 @@ def launchd_install(force: bool = False):
 
 def launchd_uninstall():
     plist_path = get_launchd_plist_path()
-    subprocess.run(["launchctl", "unload", str(plist_path)], check=False)
+    label = get_launchd_label()
+    subprocess.run(["launchctl", "bootout", f"{_launchd_domain()}/{label}"], check=False)
     
     if plist_path.exists():
         plist_path.unlink()
@@ -1025,25 +1032,25 @@ def launchd_start():
         print("↻ launchd plist missing; regenerating service definition")
         plist_path.parent.mkdir(parents=True, exist_ok=True)
         plist_path.write_text(generate_launchd_plist(), encoding="utf-8")
-        subprocess.run(["launchctl", "load", str(plist_path)], check=True)
-        subprocess.run(["launchctl", "start", label], check=True)
+        subprocess.run(["launchctl", "bootstrap", _launchd_domain(), str(plist_path)], check=True)
+        subprocess.run(["launchctl", "kickstart", f"{_launchd_domain()}/{label}"], check=True)
         print("✓ Service started")
         return
 
     refresh_launchd_plist_if_needed()
     try:
-        subprocess.run(["launchctl", "start", label], check=True)
+        subprocess.run(["launchctl", "kickstart", f"{_launchd_domain()}/{label}"], check=True)
     except subprocess.CalledProcessError as e:
         if e.returncode != 3:
             raise
         print("↻ launchd job was unloaded; reloading service definition")
-        subprocess.run(["launchctl", "load", str(plist_path)], check=True)
-        subprocess.run(["launchctl", "start", label], check=True)
+        subprocess.run(["launchctl", "bootstrap", _launchd_domain(), str(plist_path)], check=True)
+        subprocess.run(["launchctl", "kickstart", f"{_launchd_domain()}/{label}"], check=True)
     print("✓ Service started")
 
 def launchd_stop():
     label = get_launchd_label()
-    subprocess.run(["launchctl", "stop", label], check=True)
+    subprocess.run(["launchctl", "kill", "SIGTERM", f"{_launchd_domain()}/{label}"], check=True)
     print("✓ Service stopped")
 
 def _wait_for_gateway_exit(timeout: float = 10.0, force_after: float = 5.0):
@@ -1087,14 +1094,23 @@ def _wait_for_gateway_exit(timeout: float = 10.0, force_after: float = 5.0):
 
 
 def launchd_restart():
+    label = get_launchd_label()
+    target = f"{_launchd_domain()}/{label}"
+    # Use kickstart -k so launchd performs an atomic kill+restart.
+    # A two-step stop/start from inside the gateway's own process tree
+    # would kill the shell before the start command is reached.
     try:
-        launchd_stop()
+        subprocess.run(["launchctl", "kickstart", "-k", target], check=True)
+        print("✓ Service restarted")
     except subprocess.CalledProcessError as e:
         if e.returncode != 3:
             raise
-        print("↻ launchd job was unloaded; skipping stop")
-    _wait_for_gateway_exit()
-    launchd_start()
+        # Job not loaded — bootstrap and start fresh
+        print("↻ launchd job was unloaded; reloading")
+        plist_path = get_launchd_plist_path()
+        subprocess.run(["launchctl", "bootstrap", _launchd_domain(), str(plist_path)], check=True)
+        subprocess.run(["launchctl", "kickstart", target], check=True)
+        print("✓ Service restarted")
 
 def launchd_status(deep: bool = False):
     plist_path = get_launchd_plist_path()

pyproject.toml

@@ -40,7 +40,7 @@ dependencies = [
 modal = ["modal>=1.0.0,<2"]
 daytona = ["daytona>=0.148.0,<1"]
 dev = ["debugpy>=1.8.0,<2", "pytest>=9.0.2,<10", "pytest-asyncio>=1.3.0,<2", "pytest-xdist>=3.0,<4", "mcp>=1.2.0,<2"]
-messaging = ["python-telegram-bot>=22.6,<23", "discord.py[voice]>=2.7.1,<3", "aiohttp>=3.13.3,<4", "slack-bolt>=1.18.0,<2", "slack-sdk>=3.27.0,<4"]
+messaging = ["python-telegram-bot[webhooks]>=22.6,<23", "discord.py[voice]>=2.7.1,<3", "aiohttp>=3.13.3,<4", "slack-bolt>=1.18.0,<2", "slack-sdk>=3.27.0,<4"]
 cron = ["croniter>=6.0.0,<7"]
 slack = ["slack-bolt>=1.18.0,<2", "slack-sdk>=3.27.0,<4"]
 matrix = ["matrix-nio[e2e]>=0.24.0,<1", "Markdown>=3.6,<4"]

requirements.txt

@@ -31,6 +31,6 @@ edge-tts
 croniter
 
 # Optional: For messaging platform integrations (gateway)
-python-telegram-bot>=20.0
+python-telegram-bot[webhooks]>=22.6
 discord.py>=2.0
 aiohttp>=3.9.0

tests/agent/test_skill_commands.py

@@ -10,6 +10,7 @@ from agent.skill_commands import (
     build_plan_path,
     build_preloaded_skills_prompt,
     build_skill_invocation_message,
+    resolve_skill_command_key,
     scan_skill_commands,
 )
 
@@ -101,6 +102,53 @@ class TestScanSkillCommands:
         assert "/disabled-skill" not in result
 
 
+class TestResolveSkillCommandKey:
+    """Telegram bot-command names disallow hyphens, so the menu registers
+    skills with hyphens swapped for underscores. When Telegram autocomplete
+    sends the underscored form back, we need to find the hyphenated key.
+    """
+
+    def test_hyphenated_form_matches_directly(self, tmp_path):
+        with patch("tools.skills_tool.SKILLS_DIR", tmp_path):
+            _make_skill(tmp_path, "claude-code")
+            scan_skill_commands()
+            assert resolve_skill_command_key("claude-code") == "/claude-code"
+
+    def test_underscore_form_resolves_to_hyphenated_skill(self, tmp_path):
+        """/claude_code from Telegram autocomplete must resolve to /claude-code."""
+        with patch("tools.skills_tool.SKILLS_DIR", tmp_path):
+            _make_skill(tmp_path, "claude-code")
+            scan_skill_commands()
+            assert resolve_skill_command_key("claude_code") == "/claude-code"
+
+    def test_single_word_command_resolves(self, tmp_path):
+        with patch("tools.skills_tool.SKILLS_DIR", tmp_path):
+            _make_skill(tmp_path, "investigate")
+            scan_skill_commands()
+            assert resolve_skill_command_key("investigate") == "/investigate"
+
+    def test_unknown_command_returns_none(self, tmp_path):
+        with patch("tools.skills_tool.SKILLS_DIR", tmp_path):
+            _make_skill(tmp_path, "claude-code")
+            scan_skill_commands()
+            assert resolve_skill_command_key("does_not_exist") is None
+            assert resolve_skill_command_key("does-not-exist") is None
+
+    def test_empty_command_returns_none(self, tmp_path):
+        with patch("tools.skills_tool.SKILLS_DIR", tmp_path):
+            scan_skill_commands()
+            assert resolve_skill_command_key("") is None
+
+    def test_hyphenated_command_is_not_mangled(self, tmp_path):
+        """A user-typed /foo-bar (hyphen) must not trigger the underscore fallback."""
+        with patch("tools.skills_tool.SKILLS_DIR", tmp_path):
+            _make_skill(tmp_path, "foo-bar")
+            scan_skill_commands()
+            assert resolve_skill_command_key("foo-bar") == "/foo-bar"
+            # Underscore form also works (Telegram round-trip)
+            assert resolve_skill_command_key("foo_bar") == "/foo-bar"
+
+
 class TestBuildPreloadedSkillsPrompt:
     def test_builds_prompt_for_multiple_named_skills(self, tmp_path):
         with patch("tools.skills_tool.SKILLS_DIR", tmp_path):

tests/cron/test_scheduler.py

@@ -90,8 +90,9 @@ class TestResolveDeliveryTarget:
         with patch(
             "gateway.channel_directory.resolve_channel_name",
             return_value="12345678901234@lid",
-        ):
+        ) as resolve_mock:
             result = _resolve_delivery_target(job)
+        resolve_mock.assert_called_once_with("whatsapp", "Alice (dm)")
         assert result == {
             "platform": "whatsapp",
             "chat_id": "12345678901234@lid",
@@ -112,6 +113,20 @@ class TestResolveDeliveryTarget:
             "thread_id": None,
         }
 
+    def test_human_friendly_topic_label_preserves_thread_id(self):
+        """Resolved Telegram topic labels should split chat_id and thread_id."""
+        job = {"deliver": "telegram:Coaching Chat / topic 17585 (group)"}
+        with patch(
+            "gateway.channel_directory.resolve_channel_name",
+            return_value="-1009999:17585",
+        ):
+            result = _resolve_delivery_target(job)
+        assert result == {
+            "platform": "telegram",
+            "chat_id": "-1009999",
+            "thread_id": "17585",
+        }
+
     def test_raw_id_not_mangled_when_directory_returns_none(self):
         """deliver: 'whatsapp:12345@lid' passes through when directory has no match."""
         job = {"deliver": "whatsapp:12345@lid"}

tests/gateway/test_channel_directory.py

@@ -119,6 +119,19 @@ class TestResolveChannelName:
         with self._setup(tmp_path, platforms):
             assert resolve_channel_name("telegram", "Coaching Chat / topic 17585") == "-1001:17585"
 
+    def test_display_label_with_type_suffix_resolves(self, tmp_path):
+        platforms = {
+            "telegram": [
+                {"id": "123", "name": "Alice", "type": "dm"},
+                {"id": "456", "name": "Dev Group", "type": "group"},
+                {"id": "-1001:17585", "name": "Coaching Chat / topic 17585", "type": "group"},
+            ]
+        }
+        with self._setup(tmp_path, platforms):
+            assert resolve_channel_name("telegram", "Alice (dm)") == "123"
+            assert resolve_channel_name("telegram", "Dev Group (group)") == "456"
+            assert resolve_channel_name("telegram", "Coaching Chat / topic 17585 (group)") == "-1001:17585"
+
 
 class TestBuildFromSessions:
     def _write_sessions(self, tmp_path, sessions_data):

tests/gateway/test_status_command.py

@@ -128,3 +128,61 @@ async def test_handle_message_persists_agent_token_counts(monkeypatch):
         session_entry.session_key,
         last_prompt_tokens=80,
     )
+
+
+
+@pytest.mark.asyncio
+async def test_status_command_bypasses_active_session_guard():
+    """When an agent is running, /status must be dispatched immediately via
+    base.handle_message — not queued or treated as an interrupt (#5046)."""
+    import asyncio
+    from gateway.platforms.base import BasePlatformAdapter, MessageEvent, MessageType
+    from gateway.session import build_session_key
+    from gateway.config import Platform, PlatformConfig, GatewayConfig
+
+    source = _make_source()
+    session_key = build_session_key(source)
+
+    handler_called_with = []
+
+    async def fake_handler(event):
+        handler_called_with.append(event)
+        return "📊 **Hermes Gateway Status**\n**Agent Running:** Yes ⚡"
+
+    # Concrete subclass to avoid abstract method errors
+    class _ConcreteAdapter(BasePlatformAdapter):
+        platform = Platform.TELEGRAM
+
+        async def connect(self): pass
+        async def disconnect(self): pass
+        async def send(self, chat_id, content, **kwargs): pass
+        async def get_chat_info(self, chat_id): return {}
+
+    platform_config = PlatformConfig(enabled=True, token="***")
+    adapter = _ConcreteAdapter(platform_config, Platform.TELEGRAM)
+    adapter.set_message_handler(fake_handler)
+
+    sent = []
+
+    async def fake_send_with_retry(chat_id, content, reply_to=None, metadata=None):
+        sent.append(content)
+
+    adapter._send_with_retry = fake_send_with_retry
+
+    # Simulate an active session
+    interrupt_event = asyncio.Event()
+    adapter._active_sessions[session_key] = interrupt_event
+
+    event = MessageEvent(
+        text="/status",
+        source=source,
+        message_id="m1",
+        message_type=MessageType.COMMAND,
+    )
+    await adapter.handle_message(event)
+
+    assert handler_called_with, "/status handler was never called (event was queued or dropped)"
+    assert sent, "/status response was never sent"
+    assert "Agent Running" in sent[0]
+    assert not interrupt_event.is_set(), "/status incorrectly triggered an agent interrupt"
+    assert session_key not in adapter._pending_messages, "/status was incorrectly queued"

tests/gateway/test_telegram_conflict.py

@@ -80,7 +80,7 @@ async def test_polling_conflict_retries_before_fatal(monkeypatch):
         stop=AsyncMock(),
         running=True,
     )
-    bot = SimpleNamespace(set_my_commands=AsyncMock())
+    bot = SimpleNamespace(set_my_commands=AsyncMock(), delete_webhook=AsyncMock())
     app = SimpleNamespace(
         bot=bot,
         updater=updater,
@@ -99,6 +99,7 @@ async def test_polling_conflict_retries_before_fatal(monkeypatch):
     ok = await adapter.connect()
 
     assert ok is True
+    bot.delete_webhook.assert_awaited_once_with(drop_pending_updates=False)
     assert callable(captured["error_callback"])
 
     conflict = type("Conflict", (Exception,), {})
@@ -153,7 +154,7 @@ async def test_polling_conflict_becomes_fatal_after_retries(monkeypatch):
         stop=AsyncMock(),
         running=True,
     )
-    bot = SimpleNamespace(set_my_commands=AsyncMock())
+    bot = SimpleNamespace(set_my_commands=AsyncMock(), delete_webhook=AsyncMock())
     app = SimpleNamespace(
         bot=bot,
         updater=updater,
@@ -208,7 +209,7 @@ async def test_connect_marks_retryable_fatal_error_for_startup_network_failure(m
     builder = MagicMock()
     builder.token.return_value = builder
     app = SimpleNamespace(
-        bot=SimpleNamespace(),
+        bot=SimpleNamespace(delete_webhook=AsyncMock(), set_my_commands=AsyncMock()),
         updater=SimpleNamespace(),
         add_handler=MagicMock(),
         initialize=AsyncMock(side_effect=RuntimeError("Temporary failure in name resolution")),
@@ -225,6 +226,49 @@ async def test_connect_marks_retryable_fatal_error_for_startup_network_failure(m
     assert "Temporary failure in name resolution" in adapter.fatal_error_message
 
 
+@pytest.mark.asyncio
+async def test_connect_clears_webhook_before_polling(monkeypatch):
+    adapter = TelegramAdapter(PlatformConfig(enabled=True, token="***"))
+
+    monkeypatch.setattr(
+        "gateway.status.acquire_scoped_lock",
+        lambda scope, identity, metadata=None: (True, None),
+    )
+    monkeypatch.setattr(
+        "gateway.status.release_scoped_lock",
+        lambda scope, identity: None,
+    )
+
+    updater = SimpleNamespace(
+        start_polling=AsyncMock(),
+        stop=AsyncMock(),
+        running=True,
+    )
+    bot = SimpleNamespace(
+        delete_webhook=AsyncMock(),
+        set_my_commands=AsyncMock(),
+    )
+    app = SimpleNamespace(
+        bot=bot,
+        updater=updater,
+        add_handler=MagicMock(),
+        initialize=AsyncMock(),
+        start=AsyncMock(),
+    )
+    builder = MagicMock()
+    builder.token.return_value = builder
+    builder.build.return_value = app
+    monkeypatch.setattr(
+        "gateway.platforms.telegram.Application",
+        SimpleNamespace(builder=MagicMock(return_value=builder)),
+    )
+
+    ok = await adapter.connect()
+
+    assert ok is True
+    bot.delete_webhook.assert_awaited_once_with(drop_pending_updates=False)
+
+
 @pytest.mark.asyncio
 async def test_disconnect_skips_inactive_updater_and_app(monkeypatch):
     adapter = TelegramAdapter(PlatformConfig(enabled=True, token="***"))

tests/gateway/test_telegram_thread_fallback.py

@@ -37,6 +37,12 @@ class FakeTimedOut(FakeNetworkError):
     pass
 
 
+class FakeRetryAfter(Exception):
+    def __init__(self, seconds):
+        super().__init__(f"Retry after {seconds}")
+        self.retry_after = seconds
+
+
 # Build a fake telegram module tree so the adapter's internal imports work
 _fake_telegram = types.ModuleType("telegram")
 _fake_telegram_error = types.ModuleType("telegram.error")
@@ -230,3 +236,25 @@ async def test_thread_fallback_only_fires_once():
     # Second chunk: should use thread_id=None directly (effective_thread_id
     # was cleared per-chunk but the metadata doesn't change between chunks)
     # The key point: the message was delivered despite the invalid thread
+
+
+@pytest.mark.asyncio
+async def test_send_retries_retry_after_errors():
+    """Telegram flood control should back off and retry instead of failing fast."""
+    adapter = _make_adapter()
+
+    attempt = [0]
+
+    async def mock_send_message(**kwargs):
+        attempt[0] += 1
+        if attempt[0] == 1:
+            raise FakeRetryAfter(2)
+        return SimpleNamespace(message_id=300)
+
+    adapter._bot = SimpleNamespace(send_message=mock_send_message)
+
+    result = await adapter.send(chat_id="123", content="test message")
+
+    assert result.success is True
+    assert result.message_id == "300"
+    assert attempt[0] == 2

tests/gateway/test_unknown_command.py

@@ -0,0 +1,166 @@
+"""Tests for gateway warning when an unrecognized /command is dispatched.
+
+Without this warning, unknown slash commands get forwarded to the LLM as plain
+text, which often leads to silent failure (e.g. the model inventing a bogus
+delegate_task call instead of telling the user the command doesn't exist).
+"""
+
+from datetime import datetime
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from gateway.config import GatewayConfig, Platform, PlatformConfig
+from gateway.platforms.base import MessageEvent
+from gateway.session import SessionEntry, SessionSource, build_session_key
+
+
+def _make_source() -> SessionSource:
+    return SessionSource(
+        platform=Platform.TELEGRAM,
+        user_id="u1",
+        chat_id="c1",
+        user_name="tester",
+        chat_type="dm",
+    )
+
+
+def _make_event(text: str) -> MessageEvent:
+    return MessageEvent(text=text, source=_make_source(), message_id="m1")
+
+
+def _make_runner():
+    from gateway.run import GatewayRunner
+
+    runner = object.__new__(GatewayRunner)
+    runner.config = GatewayConfig(
+        platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="***")}
+    )
+    adapter = MagicMock()
+    adapter.send = AsyncMock()
+    runner.adapters = {Platform.TELEGRAM: adapter}
+    runner._voice_mode = {}
+    runner.hooks = SimpleNamespace(emit=AsyncMock(), loaded_hooks=False)
+
+    session_entry = SessionEntry(
+        session_key=build_session_key(_make_source()),
+        session_id="sess-1",
+        created_at=datetime.now(),
+        updated_at=datetime.now(),
+        platform=Platform.TELEGRAM,
+        chat_type="dm",
+    )
+    runner.session_store = MagicMock()
+    runner.session_store.get_or_create_session.return_value = session_entry
+    runner.session_store.load_transcript.return_value = []
+    runner.session_store.has_any_sessions.return_value = True
+    runner.session_store.append_to_transcript = MagicMock()
+    runner.session_store.rewrite_transcript = MagicMock()
+    runner.session_store.update_session = MagicMock()
+    runner._running_agents = {}
+    runner._pending_messages = {}
+    runner._pending_approvals = {}
+    runner._session_db = None
+    runner._reasoning_config = None
+    runner._provider_routing = {}
+    runner._fallback_model = None
+    runner._show_reasoning = False
+    runner._is_user_authorized = lambda _source: True
+    runner._set_session_env = lambda _context: None
+    runner._should_send_voice_reply = lambda *_args, **_kwargs: False
+    runner._send_voice_reply = AsyncMock()
+    runner._capture_gateway_honcho_if_configured = lambda *args, **kwargs: None
+    runner._emit_gateway_run_progress = AsyncMock()
+    return runner
+
+
+@pytest.mark.asyncio
+async def test_unknown_slash_command_returns_guidance(monkeypatch):
+    """A genuinely unknown /foobar should return user-facing guidance, not
+    silently drop through to the LLM."""
+    import gateway.run as gateway_run
+
+    runner = _make_runner()
+    # If the LLM were called, this would fail: the guard must short-circuit
+    # before _run_agent is invoked.
+    runner._run_agent = AsyncMock(
+        side_effect=AssertionError(
+            "unknown slash command leaked through to the agent"
+        )
+    )
+
+    monkeypatch.setattr(
+        gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"}
+    )
+
+    result = await runner._handle_message(_make_event("/definitely-not-a-command"))
+
+    assert result is not None
+    assert "Unknown command" in result
+    assert "/definitely-not-a-command" in result
+    assert "/commands" in result
+    runner._run_agent.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_unknown_slash_command_underscored_form_also_guarded(monkeypatch):
+    """Telegram may send /foo_bar — same guard must trigger for underscored
+    commands that normalize to unknown hyphenated names."""
+    import gateway.run as gateway_run
+
+    runner = _make_runner()
+    runner._run_agent = AsyncMock(
+        side_effect=AssertionError(
+            "unknown slash command leaked through to the agent"
+        )
+    )
+
+    monkeypatch.setattr(
+        gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"}
+    )
+
+    result = await runner._handle_message(_make_event("/made_up_thing"))
+
+    assert result is not None
+    assert "Unknown command" in result
+    assert "/made_up_thing" in result
+    runner._run_agent.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_known_slash_command_not_flagged_as_unknown(monkeypatch):
+    """A real built-in like /status must NOT hit the unknown-command guard."""
+    runner = _make_runner()
+    # Make _handle_status_command exist via the normal path by running a real
+    # dispatch. If the guard fires, the return string will mention "Unknown".
+    runner._running_agents[build_session_key(_make_source())] = MagicMock()
+
+    result = await runner._handle_message(_make_event("/status"))
+
+    assert result is not None
+    assert "Unknown command" not in result
+
+
+@pytest.mark.asyncio
+async def test_underscored_alias_for_hyphenated_builtin_not_flagged(monkeypatch):
+    """Telegram autocomplete sends /reload_mcp for the /reload-mcp built-in.
+    That must NOT be flagged as unknown."""
+    import gateway.run as gateway_run
+
+    runner = _make_runner()
+    # Prevent real MCP work; we only care that the unknown guard doesn't fire.
+    async def _noop_reload(*_a, **_kw):
+        return "mcp reloaded"
+
+    runner._handle_reload_mcp_command = _noop_reload  # type: ignore[attr-defined]
+
+    monkeypatch.setattr(
+        gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"}
+    )
+
+    result = await runner._handle_message(_make_event("/reload_mcp"))
+
+    # Whatever /reload_mcp returns, it must not be the unknown-command guard.
+    if result is not None:
+        assert "Unknown command" not in result

tests/hermes_cli/test_gateway_service.py

@@ -171,10 +171,12 @@ class TestLaunchdServiceRecovery:
 
         gateway_cli.launchd_install()
 
+        label = gateway_cli.get_launchd_label()
+        domain = gateway_cli._launchd_domain()
         assert "--replace" in plist_path.read_text(encoding="utf-8")
         assert calls[:2] == [
-            ["launchctl", "unload", str(plist_path)],
-            ["launchctl", "load", str(plist_path)],
+            ["launchctl", "bootout", f"{domain}/{label}"],
+            ["launchctl", "bootstrap", domain, str(plist_path)],
         ]
 
     def test_launchd_start_reloads_unloaded_job_and_retries(self, tmp_path, monkeypatch):
@@ -183,10 +185,12 @@ class TestLaunchdServiceRecovery:
         label = gateway_cli.get_launchd_label()
 
         calls = []
+        domain = gateway_cli._launchd_domain()
+        target = f"{domain}/{label}"
 
         def fake_run(cmd, check=False, **kwargs):
             calls.append(cmd)
-            if cmd == ["launchctl", "start", label] and calls.count(cmd) == 1:
+            if cmd == ["launchctl", "kickstart", target] and calls.count(cmd) == 1:
                 raise gateway_cli.subprocess.CalledProcessError(3, cmd, stderr="Could not find service")
             return SimpleNamespace(returncode=0, stdout="", stderr="")
 
@@ -196,9 +200,9 @@ class TestLaunchdServiceRecovery:
         gateway_cli.launchd_start()
 
         assert calls == [
-            ["launchctl", "start", label],
-            ["launchctl", "load", str(plist_path)],
-            ["launchctl", "start", label],
+            ["launchctl", "kickstart", target],
+            ["launchctl", "bootstrap", domain, str(plist_path)],
+            ["launchctl", "kickstart", target],
         ]
 
     def test_launchd_status_reports_local_stale_plist_when_unloaded(self, tmp_path, monkeypatch, capsys):
@@ -293,7 +297,7 @@ class TestGatewaySystemServiceRouting:
             gateway_cli,
             "launchd_restart",
             lambda: (_ for _ in ()).throw(
-                gateway_cli.subprocess.CalledProcessError(5, ["launchctl", "start", "ai.hermes.gateway"])
+                gateway_cli.subprocess.CalledProcessError(5, ["launchctl", "kickstart", "-k", "gui/501/ai.hermes.gateway"])
             ),
         )
 

tests/hermes_cli/test_update_gateway_restart.py

@@ -218,9 +218,9 @@ class TestLaunchdPlistRefresh:
         assert result is True
         # Plist should now contain the generated content (which includes --replace)
         assert "--replace" in plist_path.read_text()
-        # Should have unloaded then reloaded
-        assert any("unload" in str(c) for c in calls)
-        assert any("load" in str(c) for c in calls)
+        # Should have booted out then bootstrapped
+        assert any("bootout" in str(c) for c in calls)
+        assert any("bootstrap" in str(c) for c in calls)
 
     def test_refresh_skips_when_current(self, tmp_path, monkeypatch):
         plist_path = tmp_path / "ai.hermes.gateway.plist"
@@ -262,10 +262,10 @@ class TestLaunchdPlistRefresh:
 
         gateway_cli.launchd_start()
 
-        # First calls should be refresh (unload/load), then start
+        # First calls should be refresh (bootout/bootstrap), then kickstart
         cmd_strs = [" ".join(c) for c in calls]
-        assert any("unload" in s for s in cmd_strs)
-        assert any("start" in s for s in cmd_strs)
+        assert any("bootout" in s for s in cmd_strs)
+        assert any("kickstart" in s for s in cmd_strs)
 
     def test_launchd_start_recreates_missing_plist_and_loads_service(self, tmp_path, monkeypatch):
         """launchd_start self-heals when the plist file is missing entirely."""
@@ -288,11 +288,11 @@ class TestLaunchdPlistRefresh:
         assert "--replace" in plist_path.read_text()
 
         cmd_strs = [" ".join(c) for c in calls]
-        # Should load the new plist, then start
-        assert any("load" in s for s in cmd_strs)
-        assert any("start" in s for s in cmd_strs)
-        # Should NOT call unload (nothing to unload)
-        assert not any("unload" in s for s in cmd_strs)
+        # Should bootstrap the new plist, then kickstart
+        assert any("bootstrap" in s for s in cmd_strs)
+        assert any("kickstart" in s for s in cmd_strs)
+        # Should NOT call bootout (nothing to bootout)
+        assert not any("bootout" in s for s in cmd_strs)
 
 
 class TestCmdUpdateLaunchdRestart:

tests/tools/test_send_message_tool.py

@@ -203,6 +203,44 @@ class TestSendMessageTool:
             media_files=[],
         )
 
+    def test_display_label_target_resolves_via_channel_directory(self, tmp_path):
+        config, telegram_cfg = _make_config()
+        cache_file = tmp_path / "channel_directory.json"
+        cache_file.write_text(json.dumps({
+            "updated_at": "2026-01-01T00:00:00",
+            "platforms": {
+                "telegram": [
+                    {"id": "-1001:17585", "name": "Coaching Chat / topic 17585", "type": "group"}
+                ]
+            },
+        }))
+
+        with patch("gateway.channel_directory.DIRECTORY_PATH", cache_file), \
+             patch("gateway.config.load_gateway_config", return_value=config), \
+             patch("tools.interrupt.is_interrupted", return_value=False), \
+             patch("model_tools._run_async", side_effect=_run_async_immediately), \
+             patch("tools.send_message_tool._send_to_platform", new=AsyncMock(return_value={"success": True})) as send_mock, \
+             patch("gateway.mirror.mirror_to_session", return_value=True):
+            result = json.loads(
+                send_message_tool(
+                    {
+                        "action": "send",
+                        "target": "telegram:Coaching Chat / topic 17585 (group)",
+                        "message": "hello",
+                    }
+                )
+            )
+
+        assert result["success"] is True
+        send_mock.assert_awaited_once_with(
+            Platform.TELEGRAM,
+            telegram_cfg,
+            "-1001",
+            "hello",
+            thread_id="17585",
+            media_files=[],
+        )
+
     def test_media_only_message_uses_placeholder_for_mirroring(self):
         config, telegram_cfg = _make_config()
 

uv.lock

@@ -1689,7 +1689,7 @@ all = [
     { name = "pytest" },
     { name = "pytest-asyncio" },
     { name = "pytest-xdist" },
-    { name = "python-telegram-bot" },
+    { name = "python-telegram-bot", extra = ["webhooks"] },
     { name = "pywinpty", marker = "sys_platform == 'win32'" },
     { name = "simple-term-menu" },
     { name = "slack-bolt" },
@@ -1733,7 +1733,7 @@ mcp = [
 messaging = [
     { name = "aiohttp" },
     { name = "discord-py", extra = ["voice"] },
-    { name = "python-telegram-bot" },
+    { name = "python-telegram-bot", extra = ["webhooks"] },
     { name = "slack-bolt" },
     { name = "slack-sdk" },
 ]
@@ -1827,7 +1827,7 @@ requires-dist = [
     { name = "pytest-asyncio", marker = "extra == 'dev'", specifier = ">=1.3.0,<2" },
     { name = "pytest-xdist", marker = "extra == 'dev'", specifier = ">=3.0,<4" },
     { name = "python-dotenv", specifier = ">=1.2.1,<2" },
-    { name = "python-telegram-bot", marker = "extra == 'messaging'", specifier = ">=22.6,<23" },
+    { name = "python-telegram-bot", extras = ["webhooks"], marker = "extra == 'messaging'", specifier = ">=22.6,<23" },
     { name = "pywinpty", marker = "sys_platform == 'win32' and extra == 'pty'", specifier = ">=2.0.0,<3" },
     { name = "pyyaml", specifier = ">=6.0.2,<7" },
     { name = "requests", specifier = ">=2.33.0,<3" },
@@ -3964,6 +3964,11 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/13/97/7298f0e1afe3a1ae52ff4c5af5087ed4de319ea73eb3b5c8c4dd4e76e708/python_telegram_bot-22.6-py3-none-any.whl", hash = "sha256:e598fe171c3dde2dfd0f001619ee9110eece66761a677b34719fb18934935ce0", size = 737267, upload-time = "2026-01-24T13:56:58.06Z" },
 ]
 
+[package.optional-dependencies]
+webhooks = [
+    { name = "tornado" },
+]
+
 [[package]]
 name = "pytz"
 version = "2025.2"