diff --git a/cli.py b/cli.py index a22a836b7fa14..3aaa8df0ec9fb 100644 --- a/cli.py +++ b/cli.py @@ -7689,6 +7689,21 @@ def _handle_skills_command(self, cmd: str): from hermes_cli.skills_hub import handle_skills_slash handle_skills_slash(cmd, ChatConsole()) + def _handle_dashboard_command(self): + """List configured agent dashboard URLs (RFD-0002 ยง11).""" + dash_cfg = (self.config.get("dashboard") or {}) if isinstance(self.config, dict) else {} + hosts = dash_cfg.get("agent_hosts") or [] + if not hosts: + print(" No agent hosts configured. Add to ~/.hermes/config.yaml under dashboard.agent_hosts.") + return + print() + print(" Agents Dashboards:") + for host in hosts: + name = host.get("name", "?") if isinstance(host, dict) else "?" + url = host.get("url", "") if isinstance(host, dict) else "" + print(f" {name}: {url}") + print() + def _show_gateway_status(self): """Show status of the gateway and connected messaging platforms.""" from gateway.config import load_gateway_config, Platform @@ -7974,6 +7989,8 @@ def process_command(self, command: str) -> bool: self._handle_skills_command(cmd_original) elif canonical == "platforms": self._show_gateway_status() + elif canonical == "dashboard": + self._handle_dashboard_command() elif canonical == "status": self._show_session_status() elif canonical == "statusbar": diff --git a/cron/scheduler.py b/cron/scheduler.py index 6302227e8f631..1ad8cd3a65145 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -1581,6 +1581,11 @@ def _run_job_impl(job: dict) -> tuple[bool, str, str, Optional[str]]: session_id=_cron_session_id, session_db=_session_db, ) + # Cron progress is not wired to gateway adapters here: run_job() may + # execute standalone, and live adapters/loop are only available later + # in _deliver_result() for the final delivery. Keep interim Telegram + # status edits on the gateway message-processing seam until cron has a + # low-risk live-delivery callback path. # Run the agent with an *inactivity*-based timeout: the job can run # for hours if it's actively calling tools / receiving stream tokens, diff --git a/gateway/platforms/telegram.py b/gateway/platforms/telegram.py index 76288d4596d97..a7794ce099c07 100644 --- a/gateway/platforms/telegram.py +++ b/gateway/platforms/telegram.py @@ -14,6 +14,7 @@ import tempfile import html as _html import re +from collections import OrderedDict from typing import Dict, List, Optional, Any logger = logging.getLogger(__name__) @@ -330,6 +331,8 @@ class TelegramAdapter(BasePlatformAdapter): # Telegram message limits MAX_MESSAGE_LENGTH = 4096 + STATUS_MESSAGE_LENGTH = 3800 + STATUS_MESSAGE_ID_CACHE_LIMIT = 1024 # Threshold for detecting Telegram client-side message splits. # When a chunk is near this limit, a continuation is almost certain. _SPLIT_THRESHOLD = 4000 @@ -466,6 +469,14 @@ def __init__(self, config: PlatformConfig): # "all" โ€” every message triggers a push notification (legacy # behavior; opt-in via display.platforms.telegram.notifications). self._notifications_mode: str = "important" + # Status update messages sent through send_or_update_status(). Values + # are bot message IDs returned from this adapter's own send path. + self._status_message_ids: OrderedDict[ + tuple[str, Optional[str], str], str + ] = OrderedDict() + self._status_message_locks: OrderedDict[ + tuple[str, Optional[str], str], asyncio.Lock + ] = OrderedDict() def _notification_kwargs( self, metadata: Optional[Dict[str, Any]] @@ -1871,6 +1882,172 @@ async def send( is_connect_timeout = self._looks_like_connect_timeout(e) return SendResult(success=False, error=str(e), retryable=(is_connect_timeout or not is_timeout)) + @staticmethod + def _truncate_utf16_text(text: str, max_length: int, suffix: str = "...") -> str: + """Truncate text to a UTF-16 code-unit budget.""" + if utf16_len(text) <= max_length: + return text + + suffix_len = utf16_len(suffix) + budget = max(0, max_length - suffix_len) + lo, hi = 0, len(text) + best = "" + while lo <= hi: + mid = (lo + hi) // 2 + candidate = text[:mid] + if utf16_len(candidate) <= budget: + best = candidate + lo = mid + 1 + else: + hi = mid - 1 + return best.rstrip() + suffix + + def _bound_status_text(self, text: str) -> str: + """Return status text that will fit in one Telegram message.""" + raw_text = str(text or "") + bounded = self._truncate_utf16_text(raw_text, self.STATUS_MESSAGE_LENGTH) + if utf16_len(self.format_message(bounded)) <= self.STATUS_MESSAGE_LENGTH: + return bounded + + lo, hi = 0, len(raw_text) + best = "" + while lo <= hi: + mid = (lo + hi) // 2 + candidate = f"{raw_text[:mid].rstrip()}..." + fits = ( + utf16_len(candidate) <= self.STATUS_MESSAGE_LENGTH + and utf16_len(self.format_message(candidate)) <= self.STATUS_MESSAGE_LENGTH + ) + if fits: + best = candidate + lo = mid + 1 + else: + hi = mid - 1 + return best or "..." + + def _status_message_cache(self) -> OrderedDict[tuple[str, Optional[str], str], str]: + cache = getattr(self, "_status_message_ids", None) + if not isinstance(cache, OrderedDict): + cache = OrderedDict(cache or {}) + self._status_message_ids = cache + return cache + + def _get_status_message_id(self, key: tuple[str, Optional[str], str]) -> Optional[str]: + cache = self._status_message_cache() + message_id = cache.get(key) + if message_id is not None: + cache.move_to_end(key) + return message_id + + def _remember_status_message_id( + self, + key: tuple[str, Optional[str], str], + message_id: str, + ) -> None: + cache = self._status_message_cache() + cache[key] = str(message_id) + cache.move_to_end(key) + limit = max( + 1, + int(getattr(self, "STATUS_MESSAGE_ID_CACHE_LIMIT", 1024) or 1024), + ) + while len(cache) > limit: + cache.popitem(last=False) + + def _status_message_lock( + self, + key: tuple[str, Optional[str], str], + ) -> asyncio.Lock: + locks = getattr(self, "_status_message_locks", None) + if not isinstance(locks, OrderedDict): + locks = OrderedDict(locks or {}) + self._status_message_locks = locks + lock = locks.get(key) + if lock is None: + lock = asyncio.Lock() + locks[key] = lock + locks.move_to_end(key) + limit = max( + 1, + int(getattr(self, "STATUS_MESSAGE_ID_CACHE_LIMIT", 1024) or 1024), + ) + while len(locks) > limit: + oldest_key, oldest_lock = next(iter(locks.items())) + if oldest_lock.locked(): + break + locks.pop(oldest_key, None) + return lock + + @staticmethod + def _is_permanent_status_edit_failure(error: Optional[str]) -> bool: + """Return True for edit failures where replacing the status is safe.""" + err = str(error or "").lower() + permanent_markers = ( + "message to edit not found", + "message not found", + "message can't be edited", + "message cannot be edited", + "message is not editable", + "message_id_invalid", + "not enough rights to edit", + "no rights to edit", + "too old to edit", + "message is too old", + "message was deleted", + "message deleted", + "not editable", + ) + return any(marker in err for marker in permanent_markers) + + async def send_or_update_status( + self, + chat_id: str, + thread_id: Optional[str], + status_key: str, + text: str, + metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + """Send a scoped status message or edit the previous one in-place.""" + self._status_message_cache() + + bounded_text = self._bound_status_text(text) + normalized_thread_id = str(thread_id) if thread_id is not None else None + key = (str(chat_id), normalized_thread_id, str(status_key)) + status_metadata = dict(metadata) if isinstance(metadata, dict) else {} + if normalized_thread_id is not None and "thread_id" not in status_metadata: + status_metadata["thread_id"] = normalized_thread_id + metadata = status_metadata or None + + async with self._status_message_lock(key): + existing_message_id = self._get_status_message_id(key) + if existing_message_id is None: + result = await self.send(chat_id, bounded_text, metadata=metadata) + if result.success and result.message_id is not None: + self._remember_status_message_id(key, str(result.message_id)) + return result + + edit_result = await self.edit_message( + chat_id, + existing_message_id, + bounded_text, + finalize=True, + metadata=metadata, + ) + if edit_result.success: + if edit_result.message_id is not None: + self._remember_status_message_id(key, str(edit_result.message_id)) + return edit_result + + if getattr(edit_result, "retryable", False): + return edit_result + if not self._is_permanent_status_edit_failure(edit_result.error): + return edit_result + + send_result = await self.send(chat_id, bounded_text, metadata=metadata) + if send_result.success and send_result.message_id is not None: + self._remember_status_message_id(key, str(send_result.message_id)) + return send_result + async def edit_message( self, chat_id: str, diff --git a/gateway/run.py b/gateway/run.py index 7a6990c998e78..ac3c72dc734a4 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -205,6 +205,99 @@ def _prepare_gateway_status_message(platform: Any, event_type: str, message: str return text +def _bound_gateway_status_text(adapter: Any, text: str) -> str: + """Bound interim status text to one platform message before status edits.""" + status_bounder = getattr(adapter, "_bound_status_text", None) + if callable(status_bounder): + try: + return status_bounder(text) + except Exception: + pass + + raw_text = str(text or "") + try: + limit = int( + getattr( + adapter, + "STATUS_MESSAGE_LENGTH", + getattr(adapter, "MAX_MESSAGE_LENGTH", 4000), + ) + or 4000 + ) + except Exception: + limit = 4000 + limit = max(1, limit) + len_fn = getattr(adapter, "message_len_fn", len) + try: + if len_fn(raw_text) <= limit: + return raw_text + except Exception: + if len(raw_text) <= limit: + return raw_text + + suffix = "..." + budget = max(0, limit - len(suffix)) + best = "" + lo, hi = 0, len(raw_text) + while lo <= hi: + mid = (lo + hi) // 2 + candidate = raw_text[:mid] + try: + fits = len_fn(candidate) <= budget + except Exception: + fits = len(candidate) <= budget + if fits: + best = candidate + lo = mid + 1 + else: + hi = mid - 1 + return (best.rstrip() + suffix) if best else suffix + + +def _gateway_status_text_fits(adapter: Any, text: str) -> bool: + """Return True when status text already fits the adapter status bound.""" + try: + return _bound_gateway_status_text(adapter, text) == text + except Exception: + return False + + +def _bound_gateway_status_text_recent(adapter: Any, text: str) -> str: + """Bound status text while preserving the newest progress lines.""" + raw_text = str(text or "") + if not raw_text or _gateway_status_text_fits(adapter, raw_text): + return raw_text + + prefix = "..." + selected: list[str] = [] + for line in reversed(raw_text.splitlines() or [raw_text]): + body = line if not selected else line + "\n" + "\n".join(selected) + candidate = prefix + "\n" + body + if _gateway_status_text_fits(adapter, candidate): + selected.insert(0, line) + continue + + if selected: + break + + lo, hi = 0, len(line) + best = "" + while lo <= hi: + mid = (lo + hi) // 2 + suffix = line[len(line) - mid :] if mid else "" + candidate = prefix + "\n" + suffix if suffix else prefix + if _gateway_status_text_fits(adapter, candidate): + best = suffix + lo = mid + 1 + else: + hi = mid - 1 + return (prefix + "\n" + best) if best else _bound_gateway_status_text(adapter, prefix) + + if selected: + return prefix + "\n" + "\n".join(selected) + return _bound_gateway_status_text(adapter, raw_text) + + def _telegramize_command_mentions(text: str, platform: Any) -> str: """Rewrite slash-command mentions to Telegram-valid command names. @@ -1364,6 +1457,29 @@ def _preserve_queued_followup_history_offset( return merged +def _build_dashboard_keyboard(hosts): + """Build an InlineKeyboardMarkup for /dashboard. + + Returns None when ``hosts`` is empty, so callers can fall back to plain + text. Each host dict must have ``url``; ``name`` falls back to the url. + Imports ``telegram`` lazily so the gateway can run without it installed. + """ + if not hosts: + return None + from telegram import InlineKeyboardButton, InlineKeyboardMarkup + rows = [ + [InlineKeyboardButton( + text=str(h.get("name") or h.get("url")), + url=str(h.get("url")), + )] + for h in hosts + if isinstance(h, dict) and h.get("url") + ] + if not rows: + return None + return InlineKeyboardMarkup(rows) + + class GatewayRunner: """ Main gateway controller. @@ -7160,6 +7276,9 @@ async def _do_reset(): if canonical == "platform": return await self._handle_platform_command(event) + if canonical == "dashboard": + return await self._handle_dashboard_command(event) + if canonical == "restart": return await self._handle_restart_command(event) @@ -9211,6 +9330,59 @@ async def _handle_whoami_command(self, event: MessageEvent) -> str: ) + async def _handle_dashboard_command(self, event: MessageEvent) -> Optional[str]: + """Handle /dashboard โ€” list agent dashboard URLs. + + On Telegram (DM or group), emits an InlineKeyboardMarkup with one + button per host (url buttons). On other platforms, falls back to a + plain text list of URLs. Hosts read from CLI config key + ``dashboard.agent_hosts`` (RFD-0002 ยง11). + """ + from hermes_cli.config import load_config as _load_cli_config + try: + _cfg = _load_cli_config() or {} + except Exception: + _cfg = {} + dash_cfg = (_cfg.get("dashboard") or {}) if isinstance(_cfg, dict) else {} + hosts_raw = dash_cfg.get("agent_hosts") or [] + hosts: list[dict] = [h for h in hosts_raw if isinstance(h, dict) and h.get("url")] + + if not hosts: + return ( + "No agent hosts configured. Add to ~/.hermes/config.yaml under " + "`dashboard.agent_hosts` as a list of {name, url} entries." + ) + + source = event.source + is_telegram = source and source.platform == Platform.TELEGRAM + adapter = self.adapters.get(source.platform) if source else None + + if is_telegram and adapter is not None: + try: + keyboard = _build_dashboard_keyboard(hosts) + # Use the underlying bot send_message to attach reply_markup. + bot = getattr(adapter, "_bot", None) if keyboard is not None else None + if bot is not None: + try: + await bot.send_message( + chat_id=int(source.chat_id), + text="๐Ÿš€ Agents Dashboards", + reply_markup=keyboard, + ) + return None # already sent + except Exception as _e: + logger.warning("dashboard: telegram inline send failed: %s", _e) + # Fall through to text reply. + except ImportError: + pass + + # Text fallback (non-Telegram, or Telegram send failed) + lines = ["**Agents Dashboards:**"] + for h in hosts: + name = h.get("name") or h.get("url") + lines.append(f"โ€ข {name}: {h.get('url')}") + return "\n".join(lines) + async def _handle_kanban_command(self, event: MessageEvent) -> str: """Handle /kanban โ€” delegate to the shared kanban CLI. @@ -15627,6 +15799,15 @@ def progress_callback(event_type: str, tool_name: str = None, preview: str = Non if _progress_thread_id == source.thread_id else {"thread_id": _progress_thread_id} ) if _progress_thread_id else None + _status_update_thread_id = ( + _progress_thread_id + if _progress_thread_id is not None + else source.thread_id + ) + _status_update_key = ( + f"gateway-run:{session_key or session_id or source.chat_id}:" + f"{run_generation if run_generation is not None else id(progress_queue)}" + ) _progress_reply_to = ( event_message_id if source.platform in (Platform.FEISHU, Platform.MATTERMOST) and source.thread_id and event_message_id @@ -15641,10 +15822,29 @@ async def send_progress_messages(): if not adapter: return + _send_or_update_status = getattr(adapter, "send_or_update_status", None) + _status_updates_supported = callable(_send_or_update_status) + _status_update_accepts_metadata = False + if _status_updates_supported and _progress_metadata: + try: + _status_params = inspect.signature(_send_or_update_status).parameters + _status_update_accepts_metadata = ( + "metadata" in _status_params + or any( + param.kind is inspect.Parameter.VAR_KEYWORD + for param in _status_params.values() + ) + ) + except (TypeError, ValueError): + _status_update_accepts_metadata = False + # Skip tool progress for platforms that don't support message # editing (e.g. iMessage/BlueBubbles) โ€” each progress update # would become a separate message bubble, which is noisy. - if type(adapter).edit_message is BasePlatformAdapter.edit_message: + if ( + not _status_updates_supported + and type(adapter).edit_message is BasePlatformAdapter.edit_message + ): while not progress_queue.empty(): try: progress_queue.get_nowait() @@ -15727,6 +15927,20 @@ def _track_progress_result(result) -> None: _cleanup_msg_ids.append(str(result.message_id)) async def _send_progress_text(text: str): + if _status_updates_supported: + bounded_text = _bound_gateway_status_text_recent(adapter, text) + kwargs = {} + if _status_update_accepts_metadata: + kwargs["metadata"] = _progress_metadata + result = await _send_or_update_status( + source.chat_id, + _status_update_thread_id, + _status_update_key, + bounded_text, + **kwargs, + ) + _track_progress_result(result) + return result result = await adapter.send( chat_id=source.chat_id, content=text, @@ -15743,6 +15957,8 @@ async def _roll_progress_overflow_if_needed() -> bool: caller should skip the normal send/edit path for this tick. """ nonlocal progress_msg_id, progress_lines, can_edit + if _status_updates_supported: + return False if not progress_lines or not can_edit: return False groups = _split_progress_groups(progress_lines) @@ -15850,8 +16066,13 @@ async def _roll_progress_overflow_if_needed() -> bool: if can_edit and progress_msg_id is not None: # Try to edit the existing progress message full_text = "\n".join(progress_lines) - result = await _edit_progress_message(progress_msg_id, full_text) + if _status_updates_supported: + result = await _send_progress_text(full_text) + else: + result = await _edit_progress_message(progress_msg_id, full_text) if not result.success: + if _status_updates_supported: + continue _err = (getattr(result, "error", "") or "").lower() # Transient network errors (ConnectError, timeouts) # must not permanently disable progress-message @@ -15890,20 +16111,10 @@ async def _roll_progress_overflow_if_needed() -> bool: if can_edit: # First tool: send all accumulated text as new message full_text = "\n".join(progress_lines) - result = await adapter.send( - chat_id=source.chat_id, - content=full_text, - reply_to=_progress_reply_to, - metadata=_progress_metadata, - ) + result = await _send_progress_text(full_text) else: # Editing unsupported: send just this line - result = await adapter.send( - chat_id=source.chat_id, - content=msg, - reply_to=_progress_reply_to, - metadata=_progress_metadata, - ) + result = await _send_progress_text(msg) if result.success and result.message_id: progress_msg_id = result.message_id if _cleanup_progress: @@ -15936,7 +16147,10 @@ async def _roll_progress_overflow_if_needed() -> bool: if can_edit and progress_lines and progress_msg_id: _pending_text = _progress_text(progress_lines) try: - await _edit_progress_message(progress_msg_id, _pending_text) + if _status_updates_supported: + await _send_progress_text(_pending_text) + else: + await _edit_progress_message(progress_msg_id, _pending_text) except Exception: pass progress_msg_id = None @@ -15954,7 +16168,10 @@ async def _roll_progress_overflow_if_needed() -> bool: if can_edit and progress_lines and progress_msg_id: full_text = _progress_text(progress_lines) try: - await _edit_progress_message(progress_msg_id, full_text) + if _status_updates_supported: + await _send_progress_text(full_text) + else: + await _edit_progress_message(progress_msg_id, full_text) except Exception: pass return @@ -16013,6 +16230,44 @@ def _step_callback_sync(iteration: int, prev_tools: list) -> None: else: _status_thread_metadata = self._thread_metadata_for_source(source, event_message_id) if _progress_thread_id else None + async def _send_status_text(prepared_message: str): + sender = getattr(_status_adapter, "send_or_update_status", None) + if callable(sender): + bounded_message = _bound_gateway_status_text_recent(_status_adapter, prepared_message) + kwargs = {} + if _status_thread_metadata: + try: + params = inspect.signature(sender).parameters + if ( + "metadata" in params + or any( + param.kind is inspect.Parameter.VAR_KEYWORD + for param in params.values() + ) + ): + kwargs["metadata"] = _status_thread_metadata + except (TypeError, ValueError): + pass + return await sender( + _status_chat_id, + _status_update_thread_id, + _status_update_key, + bounded_message, + **kwargs, + ) + return await _status_adapter.send( + _status_chat_id, + prepared_message, + metadata=_status_thread_metadata, + ) + + async def _send_interim_assistant_text(text: str): + return await _status_adapter.send( + _status_chat_id, + text, + metadata=_status_thread_metadata, + ) + def _status_callback_sync(event_type: str, message: str) -> None: if not _status_adapter or not _run_still_current(): return @@ -16030,11 +16285,7 @@ def _status_callback_sync(event_type: str, message: str) -> None: ) return _fut = safe_schedule_threadsafe( - _status_adapter.send( - _status_chat_id, - prepared_message, - metadata=_status_thread_metadata, - ), + _send_status_text(prepared_message), _loop_for_step, logger=logger, log_message=f"status_callback ({event_type}) scheduling error", @@ -16205,11 +16456,7 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None: if already_streamed or not _status_adapter or not str(text or "").strip(): return safe_schedule_threadsafe( - _status_adapter.send( - _status_chat_id, - text, - metadata=_status_thread_metadata, - ), + _send_interim_assistant_text(text), _loop_for_step, logger=logger, log_message="interim_assistant_callback scheduling error", @@ -17566,7 +17813,7 @@ async def _notify_long_running(): and not response.get("failed") and hasattr(_cleanup_adapter, "register_post_delivery_callback") ): - _ids_snapshot = list(_cleanup_msg_ids) + _ids_snapshot = list(dict.fromkeys(_cleanup_msg_ids)) _chat_id_snapshot = source.chat_id _adapter_snapshot = _cleanup_adapter _loop_snapshot = asyncio.get_running_loop() diff --git a/hermes_cli/commands.py b/hermes_cli/commands.py index 9fc0472f51210..9a3cb1dc3ce50 100644 --- a/hermes_cli/commands.py +++ b/hermes_cli/commands.py @@ -204,6 +204,7 @@ class CommandDef: args_hint="[days]"), CommandDef("platforms", "Show gateway/messaging platform status", "Info", cli_only=True, aliases=("gateway",)), + CommandDef("dashboard", "List agent dashboard URLs (Telegram: inline buttons)", "Info"), CommandDef("platform", "Pause, resume, or list a failing gateway platform", "Info", gateway_only=True, args_hint=" [name]"), CommandDef("copy", "Copy the last assistant response to clipboard", "Info", diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 9c4197c80fbc8..b241e484f6070 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -1089,6 +1089,8 @@ def _ensure_hermes_home_managed(home: Path): # Set this to True to re-enable the surfaces with the understanding # that the numbers are a local lower-bound estimate, not billing. "show_token_analytics": False, + # List of {name, url} dicts surfaced by /dashboard slash command (RFD-0002 ยง11) + "agent_hosts": [], }, # Privacy settings diff --git a/tests/gateway/test_dashboard_command.py b/tests/gateway/test_dashboard_command.py new file mode 100644 index 0000000000000..3dd547118b91b --- /dev/null +++ b/tests/gateway/test_dashboard_command.py @@ -0,0 +1,82 @@ +"""Unit tests for the /dashboard slash command helper. + +Exercises ``gateway.run._build_dashboard_keyboard`` โ€” the pure helper that +constructs the Telegram InlineKeyboardMarkup. Avoids spinning up a real +Telegram client. + +Works against both the real python-telegram-bot package and the +MagicMock stand-in installed by tests/gateway/conftest.py: we inject +lightweight local fakes for ``InlineKeyboardButton`` / +``InlineKeyboardMarkup`` into ``sys.modules['telegram']`` for the +duration of each test so attribute round-tripping is observable. +""" +import sys +import types + +import pytest + +pytest.importorskip("telegram") + +from gateway.run import _build_dashboard_keyboard # noqa: E402 + + +class _FakeButton: + def __init__(self, *, text=None, url=None, **_): + self.text = text + self.url = url + + +class _FakeMarkup: + def __init__(self, rows): + self.inline_keyboard = tuple(tuple(r) for r in rows) + + +@pytest.fixture +def fake_telegram(monkeypatch): + """Swap telegram.InlineKeyboardButton/Markup for local fakes. + + Restored automatically via monkeypatch teardown โ€” no cross-test + pollution of ``sys.modules``. + """ + tg = sys.modules["telegram"] + monkeypatch.setattr(tg, "InlineKeyboardButton", _FakeButton, raising=False) + monkeypatch.setattr(tg, "InlineKeyboardMarkup", _FakeMarkup, raising=False) + yield + + +def _flatten(kb): + rows = getattr(kb, "inline_keyboard", None) + assert rows is not None, "keyboard missing inline_keyboard attribute" + return [btn for row in rows for btn in row] + + +def test_build_dashboard_keyboard_empty_returns_none(): + # No telegram import needed โ€” empty short-circuits. + assert _build_dashboard_keyboard([]) is None + assert _build_dashboard_keyboard(None) is None + + +def test_build_dashboard_keyboard_with_hosts(fake_telegram): + hosts = [ + {"name": "emma", "url": "https://x"}, + {"name": "i7", "url": "https://y"}, + ] + kb = _build_dashboard_keyboard(hosts) + assert isinstance(kb, _FakeMarkup) + buttons = _flatten(kb) + assert [b.text for b in buttons] == ["emma", "i7"] + assert [b.url for b in buttons] == ["https://x", "https://y"] + + +def test_build_dashboard_keyboard_skips_invalid_entries(fake_telegram): + hosts = [ + {"name": "ok", "url": "https://ok"}, + {"name": "no-url"}, # missing url -> skipped + "not-a-dict", # wrong type -> skipped + ] + kb = _build_dashboard_keyboard(hosts) + assert isinstance(kb, _FakeMarkup) + buttons = _flatten(kb) + assert len(buttons) == 1 + assert buttons[0].text == "ok" + assert buttons[0].url == "https://ok" diff --git a/tests/gateway/test_telegram_status_gateway_integration.py b/tests/gateway/test_telegram_status_gateway_integration.py new file mode 100644 index 0000000000000..906404f78bfde --- /dev/null +++ b/tests/gateway/test_telegram_status_gateway_integration.py @@ -0,0 +1,384 @@ +"""Gateway integration tests for Telegram run-scoped status updates.""" + +from __future__ import annotations + +import asyncio +import importlib +import sys +import time +import types +from types import SimpleNamespace + +import pytest + +from gateway.config import Platform, PlatformConfig +from gateway.platforms.base import BasePlatformAdapter, MessageEvent, SendResult +from gateway.session import SessionSource + + +class _StatusUpdateAgent: + def __init__(self, **kwargs): + self.tool_progress_callback = kwargs.get("tool_progress_callback") + self.status_callback = kwargs.get("status_callback") + self.tools = [] + self._interrupt_requested = False + + @property + def is_interrupted(self) -> bool: + return self._interrupt_requested + + def run_conversation(self, message, conversation_history=None, task_id=None): + self.status_callback("lifecycle", "Preparing context") + self.tool_progress_callback("tool.started", "todo", "planning", {"todos": []}) + time.sleep(0.35) + self.status_callback("lifecycle", "Still working") + return {"final_response": "done", "messages": [], "api_calls": 1} + + +class _LongProgressAgent: + def __init__(self, **kwargs): + self.tool_progress_callback = kwargs.get("tool_progress_callback") + self.status_callback = kwargs.get("status_callback") + self.tools = [] + self._interrupt_requested = False + + @property + def is_interrupted(self) -> bool: + return self._interrupt_requested + + def run_conversation(self, message, conversation_history=None, task_id=None): + for idx in range(20): + self.tool_progress_callback( + "tool.started", + "web_search", + f"old progress line {idx:02d}", + {}, + ) + self.tool_progress_callback( + "tool.started", + "web_search", + "current work: final query", + {}, + ) + time.sleep(0.45) + return {"final_response": "done", "messages": [], "api_calls": 1} + + +class _InterimAssistantAgent: + def __init__(self, **kwargs): + self.tool_progress_callback = kwargs.get("tool_progress_callback") + self.status_callback = kwargs.get("status_callback") + self.interim_assistant_callback = kwargs.get("interim_assistant_callback") + self.tools = [] + self._interrupt_requested = False + + @property + def is_interrupted(self) -> bool: + return self._interrupt_requested + + def run_conversation(self, message, conversation_history=None, task_id=None): + self.interim_assistant_callback("You're welcome.", already_streamed=False) + self.tool_progress_callback("tool.started", "todo", "planning", {"todos": []}) + self.status_callback("lifecycle", "Still working") + time.sleep(0.35) + return { + "final_response": "You're welcome.", + "response_previewed": True, + "messages": [], + "api_calls": 1, + } + + +class _BaseCaptureAdapter(BasePlatformAdapter): + def __init__(self, platform: Platform): + super().__init__(PlatformConfig(enabled=True, token="***"), platform) + self.sent: list[dict] = [] + self.edits: list[dict] = [] + self.typing: list[dict] = [] + self._next_id = 0 + + async def connect(self) -> bool: + return True + + async def disconnect(self) -> None: + return None + + async def send(self, chat_id, content, reply_to=None, metadata=None) -> SendResult: + self._next_id += 1 + message_id = f"send-{self._next_id}" + self.sent.append( + { + "chat_id": chat_id, + "content": content, + "reply_to": reply_to, + "metadata": metadata, + "message_id": message_id, + } + ) + return SendResult(success=True, message_id=message_id) + + async def edit_message(self, chat_id, message_id, content, *, metadata=None) -> SendResult: + self.edits.append( + { + "chat_id": chat_id, + "message_id": message_id, + "content": content, + "metadata": metadata, + } + ) + return SendResult(success=True, message_id=message_id) + + async def send_typing(self, chat_id, metadata=None) -> None: + self.typing.append({"chat_id": chat_id, "metadata": metadata}) + + async def stop_typing(self, chat_id) -> None: + return None + + async def get_chat_info(self, chat_id: str): + return {"id": chat_id} + + +class _TelegramStatusAdapter(_BaseCaptureAdapter): + def __init__(self): + super().__init__(Platform.TELEGRAM) + self.status_updates: list[dict] = [] + self.deleted: list[dict] = [] + + async def send_or_update_status( + self, + chat_id, + thread_id, + status_key, + text, + metadata=None, + ) -> SendResult: + self.status_updates.append( + { + "chat_id": chat_id, + "thread_id": thread_id, + "status_key": status_key, + "text": text, + "metadata": metadata, + } + ) + return SendResult(success=True, message_id="status-1") + + async def delete_message(self, chat_id: str, message_id: str) -> bool: + self.deleted.append({"chat_id": chat_id, "message_id": message_id}) + return True + + +class _SlackEditableAdapter(_BaseCaptureAdapter): + def __init__(self): + super().__init__(Platform.SLACK) + + +def _make_runner(adapter): + gateway_run = importlib.import_module("gateway.run") + runner = object.__new__(gateway_run.GatewayRunner) + runner.adapters = {adapter.platform: adapter} + runner._voice_mode = {} + runner._prefill_messages = [] + runner._ephemeral_system_prompt = "" + runner._reasoning_config = None + runner._session_reasoning_overrides = {} + runner._provider_routing = {} + runner._fallback_model = None + runner._session_db = None + runner._running_agents = {} + runner._running_agents_ts = {} + runner._busy_ack_ts = {} + runner._session_run_generation = {} + runner._agent_cache = {} + runner._agent_cache_lock = None + runner._session_model_overrides = {} + runner.hooks = SimpleNamespace(loaded_hooks=False) + runner.config = SimpleNamespace( + thread_sessions_per_user=False, + group_sessions_per_user=False, + stt_enabled=False, + streaming=SimpleNamespace(enabled=False, transport="off"), + ) + return runner + + +async def _run_gateway_agent( + monkeypatch, + tmp_path, + adapter, + platform: Platform, + *, + cleanup_progress: bool = False, + agent_cls=_StatusUpdateAgent, +): + monkeypatch.setenv("HERMES_TOOL_PROGRESS_MODE", "all") + + fake_dotenv = types.ModuleType("dotenv") + fake_dotenv.load_dotenv = lambda *args, **kwargs: None + monkeypatch.setitem(sys.modules, "dotenv", fake_dotenv) + + fake_run_agent = types.ModuleType("run_agent") + fake_run_agent.AIAgent = agent_cls + monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent) + + gateway_run = importlib.import_module("gateway.run") + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + display_config = ( + {"platforms": {platform.value: {"cleanup_progress": True}}} + if cleanup_progress + else {} + ) + monkeypatch.setattr( + gateway_run, + "_load_gateway_config", + lambda: {"display": display_config}, + ) + monkeypatch.setattr( + gateway_run, + "_resolve_runtime_agent_kwargs", + lambda: {"api_key": "fake", "model": "fake-model"}, + ) + + runner = _make_runner(adapter) + session_key = f"agent:main:{platform.value}:group:chat-1:thread-1" + runner._session_run_generation[session_key] = 7 + source = SessionSource( + platform=platform, + chat_id="chat-1", + chat_type="group", + thread_id="thread-1", + ) + result = await runner._run_agent( + message="hi", + context_prompt="", + history=[], + source=source, + session_id="session-1", + session_key=session_key, + run_generation=7, + event_message_id="msg-1", + ) + return result + + +@pytest.mark.asyncio +async def test_telegram_progress_and_status_share_one_status_key_without_normal_sends(monkeypatch, tmp_path): + adapter = _TelegramStatusAdapter() + + result = await _run_gateway_agent(monkeypatch, tmp_path, adapter, Platform.TELEGRAM) + + assert result["final_response"] == "done" + assert len(adapter.status_updates) >= 2 + assert {call["status_key"] for call in adapter.status_updates} == { + "gateway-run:agent:main:telegram:group:chat-1:thread-1:7" + } + assert {call["thread_id"] for call in adapter.status_updates} == {"thread-1"} + assert all(call["metadata"]["thread_id"] == "thread-1" for call in adapter.status_updates) + assert adapter.sent == [] + assert adapter.edits == [] + + +@pytest.mark.asyncio +async def test_telegram_interim_assistant_uses_normal_send_not_status_update(monkeypatch, tmp_path): + adapter = _TelegramStatusAdapter() + + result = await _run_gateway_agent( + monkeypatch, + tmp_path, + adapter, + Platform.TELEGRAM, + agent_cls=_InterimAssistantAgent, + ) + + assert result["final_response"] == "You're welcome." + assert result.get("already_sent") is True + assert [call["content"] for call in adapter.sent] == ["You're welcome."] + assert all(call["text"] != "You're welcome." for call in adapter.status_updates) + assert adapter.status_updates + assert {call["status_key"] for call in adapter.status_updates} == { + "gateway-run:agent:main:telegram:group:chat-1:thread-1:7" + } + + +@pytest.mark.asyncio +async def test_telegram_status_cleanup_deletes_repeated_status_message_once(monkeypatch, tmp_path): + adapter = _TelegramStatusAdapter() + session_key = "agent:main:telegram:group:chat-1:thread-1" + + result = await _run_gateway_agent( + monkeypatch, + tmp_path, + adapter, + Platform.TELEGRAM, + cleanup_progress=True, + ) + + assert result["final_response"] == "done" + assert len(adapter.status_updates) >= 2 + cb = adapter.pop_post_delivery_callback(session_key) + assert callable(cb) + cb() + for _ in range(20): + await asyncio.sleep(0.01) + if adapter.deleted: + break + + assert adapter.deleted == [{"chat_id": "chat-1", "message_id": "status-1"}] + + +@pytest.mark.asyncio +async def test_telegram_status_update_progress_preserves_latest_line(monkeypatch, tmp_path): + adapter = _TelegramStatusAdapter() + adapter.STATUS_MESSAGE_LENGTH = 120 + + result = await _run_gateway_agent( + monkeypatch, + tmp_path, + adapter, + Platform.TELEGRAM, + agent_cls=_LongProgressAgent, + ) + + assert result["final_response"] == "done" + assert adapter.status_updates + latest_render = adapter.status_updates[-1]["text"] + assert "current work: final query" in latest_render + assert "old progress line 00" not in latest_render + assert len(latest_render) <= adapter.STATUS_MESSAGE_LENGTH + + +@pytest.mark.asyncio +async def test_non_telegram_progress_and_status_still_use_normal_send(monkeypatch, tmp_path): + adapter = _SlackEditableAdapter() + + result = await _run_gateway_agent(monkeypatch, tmp_path, adapter, Platform.SLACK) + + assert result["final_response"] == "done" + rendered = "\n".join(call["content"] for call in adapter.sent) + assert "Preparing context" in rendered + assert "Still working" in rendered + assert any("todo" in call["content"] for call in adapter.sent + adapter.edits) + + +@pytest.mark.asyncio +async def test_final_response_delivery_remains_normal_send_new_behavior(): + adapter = _TelegramStatusAdapter() + + async def handler(_event): + return "Final response" + + adapter.set_message_handler(handler) + source = SessionSource( + platform=Platform.TELEGRAM, + chat_id="chat-1", + chat_type="group", + thread_id="thread-1", + ) + event = MessageEvent(text="hi", source=source, message_id="msg-1") + + await adapter._process_message_background(event, "session-key") + + assert adapter.status_updates == [] + assert [call["content"] for call in adapter.sent] == ["Final response"] + assert adapter.sent[0]["metadata"]["thread_id"] == "thread-1" + assert adapter.sent[0]["metadata"]["notify"] is True diff --git a/tests/gateway/test_telegram_status_update.py b/tests/gateway/test_telegram_status_update.py new file mode 100644 index 0000000000000..172f210ecc125 --- /dev/null +++ b/tests/gateway/test_telegram_status_update.py @@ -0,0 +1,270 @@ +"""Tests for Telegram status send-or-edit storage.""" + +from __future__ import annotations + +import asyncio +import sys +import types +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from gateway.config import PlatformConfig +from gateway.platforms.base import utf16_len + + +class FakeNetworkError(Exception): + pass + + +class FakeBadRequest(FakeNetworkError): + pass + + +class FakeTimedOut(FakeNetworkError): + pass + + +def _install_fake_telegram(monkeypatch): + fake_telegram = types.ModuleType("telegram") + fake_telegram.Update = SimpleNamespace(ALL_TYPES=()) + fake_telegram.Bot = object + fake_telegram.Message = object + fake_telegram.InlineKeyboardButton = object + fake_telegram.InlineKeyboardMarkup = object + + fake_error = types.ModuleType("telegram.error") + fake_error.NetworkError = FakeNetworkError + fake_error.BadRequest = FakeBadRequest + fake_error.TimedOut = FakeTimedOut + fake_telegram.error = fake_error + + fake_constants = types.ModuleType("telegram.constants") + fake_constants.ParseMode = SimpleNamespace(MARKDOWN_V2="MarkdownV2") + fake_constants.ChatType = SimpleNamespace( + GROUP="group", + SUPERGROUP="supergroup", + CHANNEL="channel", + PRIVATE="private", + ) + fake_telegram.constants = fake_constants + + fake_ext = types.ModuleType("telegram.ext") + fake_ext.Application = object + fake_ext.CommandHandler = object + fake_ext.CallbackQueryHandler = object + fake_ext.MessageHandler = object + fake_ext.ContextTypes = SimpleNamespace(DEFAULT_TYPE=object) + fake_ext.filters = object + + fake_request = types.ModuleType("telegram.request") + fake_request.HTTPXRequest = object + + monkeypatch.setitem(sys.modules, "telegram", fake_telegram) + monkeypatch.setitem(sys.modules, "telegram.error", fake_error) + monkeypatch.setitem(sys.modules, "telegram.constants", fake_constants) + monkeypatch.setitem(sys.modules, "telegram.ext", fake_ext) + monkeypatch.setitem(sys.modules, "telegram.request", fake_request) + + +@pytest.fixture +def adapter(monkeypatch): + _install_fake_telegram(monkeypatch) + from gateway.platforms.telegram import TelegramAdapter + + adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token")) + adapter._bot = MagicMock() + adapter._bot.send_chat_action = AsyncMock() + return adapter + + +@pytest.mark.asyncio +async def test_first_send_stores_mapping_and_returns_message_id(adapter): + adapter._bot.send_message = AsyncMock(return_value=SimpleNamespace(message_id=101)) + + result = await adapter.send_or_update_status("123", "77", "cron", "Working") + + assert result.success is True + assert result.message_id == "101" + assert adapter._status_message_ids[("123", "77", "cron")] == "101" + adapter._bot.send_message.assert_awaited_once() + kwargs = adapter._bot.send_message.await_args.kwargs + assert kwargs["chat_id"] == 123 + assert kwargs["message_thread_id"] == 77 + + +@pytest.mark.asyncio +async def test_second_call_edits_prior_message_without_new_send(adapter): + adapter._bot.send_message = AsyncMock(return_value=SimpleNamespace(message_id=101)) + adapter._bot.edit_message_text = AsyncMock() + + await adapter.send_or_update_status("123", "77", "cron", "Starting") + adapter._bot.send_message.reset_mock() + result = await adapter.send_or_update_status("123", "77", "cron", "Still working") + + assert result.success is True + assert result.message_id == "101" + adapter._bot.send_message.assert_not_awaited() + adapter._bot.edit_message_text.assert_awaited_once() + kwargs = adapter._bot.edit_message_text.await_args.kwargs + assert kwargs["chat_id"] == 123 + assert kwargs["message_id"] == 101 + assert kwargs["text"] == "Still working" + + +@pytest.mark.asyncio +async def test_concurrent_same_key_status_updates_do_not_duplicate_send(adapter): + send_started = asyncio.Event() + allow_send = asyncio.Event() + + async def fake_send(**kwargs): + send_started.set() + await allow_send.wait() + return SimpleNamespace(message_id=101) + + adapter._bot.send_message = AsyncMock(side_effect=fake_send) + adapter._bot.edit_message_text = AsyncMock() + + first = asyncio.create_task( + adapter.send_or_update_status("123", "77", "cron", "Starting") + ) + await asyncio.wait_for(send_started.wait(), timeout=1) + + second = asyncio.create_task( + adapter.send_or_update_status("123", "77", "cron", "Still working") + ) + await asyncio.sleep(0) + + assert adapter._bot.send_message.await_count == 1 + + allow_send.set() + first_result, second_result = await asyncio.gather(first, second) + + assert first_result.success is True + assert second_result.success is True + assert adapter._bot.send_message.await_count == 1 + adapter._bot.edit_message_text.assert_awaited_once() + kwargs = adapter._bot.edit_message_text.await_args.kwargs + assert kwargs["message_id"] == 101 + assert kwargs["text"] == "Still working" + assert adapter._status_message_ids[("123", "77", "cron")] == "101" + + +@pytest.mark.asyncio +async def test_permanent_edit_failure_sends_new_and_refreshes_mapping(adapter): + adapter._bot.send_message = AsyncMock( + side_effect=[ + SimpleNamespace(message_id=101), + SimpleNamespace(message_id=202), + ] + ) + adapter._bot.edit_message_text = AsyncMock( + side_effect=FakeBadRequest("Bad Request: message to edit not found") + ) + + await adapter.send_or_update_status("123", "77", "cron", "Starting") + result = await adapter.send_or_update_status("123", "77", "cron", "Replacement") + + assert result.success is True + assert result.message_id == "202" + assert adapter._status_message_ids[("123", "77", "cron")] == "202" + assert adapter._bot.send_message.await_count == 2 + + +@pytest.mark.asyncio +async def test_transient_edit_failure_does_not_duplicate_or_refresh(adapter): + adapter._bot.send_message = AsyncMock(return_value=SimpleNamespace(message_id=101)) + adapter._bot.edit_message_text = AsyncMock( + side_effect=FakeTimedOut("Timed out waiting for Telegram response") + ) + + await adapter.send_or_update_status("123", "77", "cron", "Starting") + adapter._bot.send_message.reset_mock() + result = await adapter.send_or_update_status("123", "77", "cron", "Still working") + + assert result.success is False + assert result.retryable is True + assert "Timed out" in result.error + assert adapter._status_message_ids[("123", "77", "cron")] == "101" + adapter._bot.send_message.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_thread_and_status_key_scoping_create_separate_messages(adapter): + next_id = 100 + + async def fake_send(**kwargs): + nonlocal next_id + next_id += 1 + return SimpleNamespace(message_id=next_id) + + adapter._bot.send_message = AsyncMock(side_effect=fake_send) + + await adapter.send_or_update_status("123", "1", "cron", "A") + await adapter.send_or_update_status("123", "2", "cron", "B") + await adapter.send_or_update_status("123", "1", "other", "C") + + assert adapter._bot.send_message.await_count == 3 + assert adapter._status_message_ids[("123", "1", "cron")] == "101" + assert adapter._status_message_ids[("123", "2", "cron")] == "102" + assert adapter._status_message_ids[("123", "1", "other")] == "103" + + +@pytest.mark.asyncio +async def test_status_message_id_cache_is_bounded_and_fallback_refreshes_recency(adapter): + adapter.STATUS_MESSAGE_ID_CACHE_LIMIT = 2 + adapter._bot.send_message = AsyncMock( + side_effect=[ + SimpleNamespace(message_id=101), + SimpleNamespace(message_id=102), + SimpleNamespace(message_id=111), + SimpleNamespace(message_id=103), + ] + ) + adapter._bot.edit_message_text = AsyncMock( + side_effect=FakeBadRequest("Bad Request: message to edit not found") + ) + + await adapter.send_or_update_status("123", "1", "first", "A") + await adapter.send_or_update_status("123", "2", "second", "B") + await adapter.send_or_update_status("123", "1", "first", "A replacement") + await adapter.send_or_update_status("123", "3", "third", "C") + + assert ("123", "2", "second") not in adapter._status_message_ids + assert adapter._status_message_ids[("123", "1", "first")] == "111" + assert adapter._status_message_ids[("123", "3", "third")] == "103" + assert len(adapter._status_message_ids) == 2 + + +@pytest.mark.asyncio +async def test_long_status_text_is_bounded_to_single_safe_message(adapter): + sent_texts = [] + + async def fake_send(**kwargs): + sent_texts.append(kwargs["text"]) + return SimpleNamespace(message_id=101) + + adapter._bot.send_message = AsyncMock(side_effect=fake_send) + long_text = "x" * 6000 + + result = await adapter.send_or_update_status("123", None, "cron", long_text) + + assert result.success is True + assert adapter._bot.send_message.await_count == 1 + assert len(sent_texts) == 1 + assert utf16_len(sent_texts[0]) <= adapter.STATUS_MESSAGE_LENGTH + assert sent_texts[0].endswith(r"\.\.\.") + + +@pytest.mark.asyncio +async def test_normal_send_behavior_remains_unaffected(adapter): + adapter._bot.send_message = AsyncMock(return_value=SimpleNamespace(message_id=303)) + + result = await adapter.send("123", "Normal send", metadata={"thread_id": "77"}) + + assert result.success is True + assert result.message_id == "303" + assert adapter._status_message_ids == {} + kwargs = adapter._bot.send_message.await_args.kwargs + assert kwargs["message_thread_id"] == 77 diff --git a/tests/hermes_cli/test_commands.py b/tests/hermes_cli/test_commands.py index 6de778347e13c..679fff099c11f 100644 --- a/tests/hermes_cli/test_commands.py +++ b/tests/hermes_cli/test_commands.py @@ -87,6 +87,13 @@ def test_cli_only_and_gateway_only_are_mutually_exclusive(self): assert not (cmd.cli_only and cmd.gateway_only), \ f"{cmd.name} cannot be both cli_only and gateway_only" + def test_dashboard_command_registered(self): + names = [c.name for c in COMMAND_REGISTRY] + assert "dashboard" in names, "dashboard command missing from registry" + cmd = next(c for c in COMMAND_REGISTRY if c.name == "dashboard") + assert cmd.category == "Info" + assert resolve_command("dashboard").name == "dashboard" + # --------------------------------------------------------------------------- # resolve_command tests