fix(status): surface degraded gateway health for Telegram runtime failures

gateway/platforms/telegram.py

@@ -184,6 +184,19 @@ def _looks_like_network_error(error: Exception) -> bool:
             pass
         return isinstance(error, OSError)
 
+    def _mark_polling_reconnecting(self, code: str, message: str) -> None:
+        """Persist that Telegram is degraded and currently trying to recover polling."""
+        try:
+            from gateway.status import write_runtime_status
+            write_runtime_status(
+                platform=self.platform.value,
+                platform_state="reconnecting",
+                error_code=code,
+                error_message=message,
+            )
+        except Exception:
+            pass
+
     async def _handle_polling_network_error(self, error: Exception) -> None:
         """Reconnect polling after a transient network interruption.
 
@@ -205,6 +218,10 @@ async def _handle_polling_network_error(self, error: Exception) -> None:
 
         self._polling_network_error_count += 1
         attempt = self._polling_network_error_count
+        self._mark_polling_reconnecting(
+            "telegram_polling_reconnect",
+            f"Telegram polling reconnect attempt {attempt}/{MAX_NETWORK_RETRIES}: {error}",
+        )
 
         if attempt > MAX_NETWORK_RETRIES:
             message = (
@@ -240,7 +257,12 @@ async def _handle_polling_network_error(self, error: Exception) -> None:
                 self.name, attempt,
             )
             self._polling_network_error_count = 0
+            self._mark_connected()
         except Exception as retry_err:
+            self._mark_polling_reconnecting(
+                "telegram_polling_reconnect",
+                f"Telegram polling reconnect failed on attempt {attempt}/{MAX_NETWORK_RETRIES}: {retry_err}",
+            )
             logger.warning("[%s] Telegram polling reconnect failed: %s", self.name, retry_err)
             # start_polling failed — polling is dead and no further error
             # callbacks will fire, so schedule the next retry ourselves.
@@ -265,6 +287,13 @@ async def _handle_polling_conflict(self, error: Exception) -> None:
         RETRY_DELAY = 10  # seconds
 
         if self._polling_conflict_count <= MAX_CONFLICT_RETRIES:
+            self._mark_polling_reconnecting(
+                "telegram_polling_conflict_retry",
+                (
+                    "Telegram polling conflict retry "
+                    f"{self._polling_conflict_count}/{MAX_CONFLICT_RETRIES}: {error}"
+                ),
+            )
             logger.warning(
                 "[%s] Telegram polling conflict (%d/%d), will retry in %ds. Error: %s",
                 self.name, self._polling_conflict_count, MAX_CONFLICT_RETRIES,
@@ -284,8 +313,16 @@ async def _handle_polling_conflict(self, error: Exception) -> None:
                 )
                 logger.info("[%s] Telegram polling resumed after conflict retry %d", self.name, self._polling_conflict_count)
                 self._polling_conflict_count = 0  # reset on success
+                self._mark_connected()
                 return
             except Exception as retry_err:
+                self._mark_polling_reconnecting(
+                    "telegram_polling_conflict_retry",
+                    (
+                        "Telegram polling conflict retry failed "
+                        f"{self._polling_conflict_count}/{MAX_CONFLICT_RETRIES}: {retry_err}"
+                    ),
+                )
                 logger.warning("[%s] Telegram polling retry failed: %s", self.name, retry_err)
                 # Don't fall through to fatal yet — wait for the next conflict
                 # to trigger another retry attempt (up to MAX_CONFLICT_RETRIES).

gateway/run.py

@@ -1062,7 +1062,16 @@ async def start(self) -> bool:
             pass
         try:
             from gateway.status import write_runtime_status
-            write_runtime_status(gateway_state="starting", exit_reason=None)
+            write_runtime_status(
+                gateway_state="starting",
+                exit_reason=None,
+                reset_platforms=True,
+                known_platforms=[
+                    platform.value
+                    for platform, platform_config in self.config.platforms.items()
+                    if getattr(platform_config, "enabled", False)
+                ],
+            )
         except Exception:
             pass
 

gateway/status.py

@@ -192,6 +192,8 @@ def write_runtime_status(
     platform_state: Optional[str] = None,
     error_code: Optional[str] = None,
     error_message: Optional[str] = None,
+    reset_platforms: bool = False,
+    known_platforms: Optional[list[str]] = None,
 ) -> None:
     """Persist gateway runtime health information for diagnostics/status."""
     path = _get_runtime_status_path()
@@ -202,6 +204,11 @@ def write_runtime_status(
     payload["start_time"] = _get_process_start_time(os.getpid())
     payload["updated_at"] = _utc_now_iso()
 
+    if reset_platforms:
+        payload["platforms"] = {}
+    if known_platforms is not None:
+        payload["known_platforms"] = list(dict.fromkeys(str(p) for p in known_platforms if p))
+
     if gateway_state is not None:
         payload["gateway_state"] = gateway_state
     if exit_reason is not None:
@@ -226,6 +233,99 @@ def read_runtime_status() -> Optional[dict[str, Any]]:
     return _read_json_file(_get_runtime_status_path())
 
 
+_NON_DEGRADING_PLATFORM_STATES = {"disconnected"}
+_DEFAULT_IGNORED_RUNTIME_PLATFORMS = {"api_server", "webhook", "local"}
+
+
+def _effective_runtime_platforms(
+    state: Optional[dict[str, Any]],
+    relevant_platforms: Optional[set[str]] = None,
+) -> dict[str, Any]:
+    if not state:
+        return {}
+
+    platforms = state.get("platforms", {}) or {}
+    if relevant_platforms is None:
+        known = state.get("known_platforms")
+        if isinstance(known, list) and known:
+            relevant_platforms = {str(item) for item in known if item}
+
+    if relevant_platforms is None:
+        return {
+            platform: pdata
+            for platform, pdata in platforms.items()
+            if platform not in _DEFAULT_IGNORED_RUNTIME_PLATFORMS
+        }
+
+    return {
+        platform: pdata
+        for platform, pdata in platforms.items()
+        if platform in relevant_platforms and platform not in _DEFAULT_IGNORED_RUNTIME_PLATFORMS
+    }
+
+
+def iter_runtime_issue_lines(
+    state: Optional[dict[str, Any]],
+    *,
+    relevant_platforms: Optional[set[str]] = None,
+    include_disconnected: bool = False,
+) -> list[str]:
+    """Return human-readable issue lines for degraded runtime platform states."""
+    if not state:
+        return []
+
+    lines: list[str] = []
+    platforms = _effective_runtime_platforms(state, relevant_platforms)
+    for platform, pdata in platforms.items():
+        platform_state = str(pdata.get("state") or "").strip().lower()
+        message = str(pdata.get("error_message") or "").strip()
+        if platform_state == "fatal":
+            lines.append(f"{platform}: {message or 'fatal error'}")
+        elif platform_state == "reconnecting":
+            lines.append(f"{platform}: reconnecting — {message or 'recovery in progress'}")
+        elif platform_state == "disconnected" and include_disconnected:
+            lines.append(f"{platform}: disconnected")
+
+    gateway_state = str(state.get("gateway_state") or "").strip().lower()
+    exit_reason = str(state.get("exit_reason") or "").strip()
+    if gateway_state == "startup_failed" and exit_reason:
+        lines.append(f"Last startup issue: {exit_reason}")
+    elif gateway_state == "stopped" and exit_reason:
+        lines.append(f"Last shutdown reason: {exit_reason}")
+
+    return lines
+
+
+def runtime_health_level(
+    state: Optional[dict[str, Any]],
+    *,
+    relevant_platforms: Optional[set[str]] = None,
+    include_disconnected: bool = False,
+) -> str:
+    """Classify runtime health as healthy, degraded, failed, or unknown."""
+    if not state:
+        return "unknown"
+
+    gateway_state = str(state.get("gateway_state") or "").strip().lower()
+    if gateway_state == "startup_failed":
+        return "failed"
+    if gateway_state == "stopped":
+        return "failed"
+
+    for pdata in _effective_runtime_platforms(state, relevant_platforms).values():
+        platform_state = str((pdata or {}).get("state") or "").strip().lower()
+        if platform_state == "fatal":
+            return "failed"
+        if platform_state == "reconnecting":
+            return "degraded"
+        if include_disconnected and platform_state == "disconnected":
+            return "degraded"
+
+    if gateway_state == "running":
+        return "healthy"
+    return "unknown"
+
+
 def remove_pid_file() -> None:
     """Remove the gateway PID file if it exists."""
     try:

hermes_cli/gateway.py

@@ -14,7 +14,7 @@
 
 PROJECT_ROOT = Path(__file__).parent.parent.resolve()
 
-from hermes_cli.config import get_env_value, get_hermes_home, save_env_value, is_managed, managed_error
+from hermes_cli.config import get_env_value, get_hermes_home, save_env_value, is_managed, managed_error, load_config
 # display_hermes_home is imported lazily at call sites to avoid ImportError
 # when hermes_constants is cached from a pre-update version during `hermes update`.
 from hermes_cli.setup import (
@@ -923,7 +923,7 @@ def systemd_status(deep: bool = False, system: bool = False):
         print()
         print("Recent gateway health:")
         for line in runtime_lines:
-            print(f"  {line}")
+            print(f"  {color('⚠', Colors.YELLOW)} {line}")
 
     if system:
         print("✓ System service starts at boot without requiring systemd linger")
@@ -1598,30 +1598,27 @@ def _platform_status(platform: dict) -> str:
 def _runtime_health_lines() -> list[str]:
     """Summarize the latest persisted gateway runtime health state."""
     try:
-        from gateway.status import read_runtime_status
+        from gateway.status import iter_runtime_issue_lines, read_runtime_status
     except Exception:
         return []
 
-    state = read_runtime_status()
-    if not state:
-        return []
-
-    lines: list[str] = []
-    gateway_state = state.get("gateway_state")
-    exit_reason = state.get("exit_reason")
-    platforms = state.get("platforms", {}) or {}
-
-    for platform, pdata in platforms.items():
-        if pdata.get("state") == "fatal":
-            message = pdata.get("error_message") or "unknown error"
-            lines.append(f"⚠ {platform}: {message}")
-
-    if gateway_state == "startup_failed" and exit_reason:
-        lines.append(f"⚠ Last startup issue: {exit_reason}")
-    elif gateway_state == "stopped" and exit_reason:
-        lines.append(f"⚠ Last shutdown reason: {exit_reason}")
+    relevant_platforms = None
+    try:
+        cfg = load_config() or {}
+        platform_cfg = cfg.get("platforms", {}) if isinstance(cfg, dict) else {}
+        relevant_platforms = {
+            str(name)
+            for name, pdata in platform_cfg.items()
+            if isinstance(pdata, dict) and pdata.get("enabled", True)
+        }
+    except Exception:
+        relevant_platforms = None
 
-    return lines
+    return iter_runtime_issue_lines(
+        read_runtime_status(),
+        relevant_platforms=relevant_platforms,
+        include_disconnected=False,
+    )
 
 
 def _setup_standard_platform(platform: dict):
@@ -2217,7 +2214,7 @@ def gateway_command(args):
                     print()
                     print("Recent gateway health:")
                     for line in runtime_lines:
-                        print(f"  {line}")
+                        print(f"  {color('⚠', Colors.YELLOW)} {line}")
                 print()
                 print("To install as a service:")
                 print("  hermes gateway install")
@@ -2229,7 +2226,7 @@ def gateway_command(args):
                     print()
                     print("Recent gateway health:")
                     for line in runtime_lines:
-                        print(f"  {line}")
+                        print(f"  {color('⚠', Colors.YELLOW)} {line}")
                 print()
                 print("To start:")
                 print("  hermes gateway          # Run in foreground")