diff --git a/gateway/run.py b/gateway/run.py index a3c36ea193108..cef8dc845fa6a 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -5978,10 +5978,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew active_count = len(getattr(self, "_running_agents", {})) if active_count < max_sessions: return None - return ( - f"Hermes is at the active session limit ({active_count}/{max_sessions}). " - "Try again when another session finishes." - ) + from hermes_cli.active_sessions import active_session_limit_message + + return active_session_limit_message(active_count, max_sessions) def _claim_active_session_slot( self, diff --git a/hermes_cli/status.py b/hermes_cli/status.py index 7537b85a1d924..4a1b0e123793d 100644 --- a/hermes_cli/status.py +++ b/hermes_cli/status.py @@ -6,6 +6,7 @@ Shows the status of all Hermes Agent components. import os import sys +import time import subprocess # noqa: F401 — re-exported for tests that monkeypatch status.subprocess to guard against regressions from pathlib import Path @@ -581,6 +582,40 @@ def show_status(args): else: print(f" Active: {_session_count if _session_count is not None else 0}") + # Slot usage, only when max_concurrent_sessions is set. The cap is shared + # across CLI, desktop/TUI and the messaging gateway, so the surface that + # gets rejected is rarely the one holding the slots — without this the only + # way to find out is reading runtime/active_sessions.json by hand. + try: + from hermes_cli.active_sessions import ( + active_session_registry_snapshot, + format_age, + resolve_max_concurrent_sessions, + ) + + _cap = resolve_max_concurrent_sessions(config) + except Exception: + _cap = None + if _cap: + try: + _held = active_session_registry_snapshot() + except Exception: + _held = [] + _full = len(_held) >= _cap + print( + " Slots: " + + color( + f"{len(_held)}/{_cap} in use", Colors.YELLOW if _full else Colors.GREEN + ) + ) + _now = time.time() + for _entry in sorted(_held, key=lambda e: e.get("started_at") or 0): + _age = format_age(_now - float(_entry.get("started_at") or _now)) + print( + f" {_entry.get('surface') or 'unknown':<17} " + f"{_entry.get('session_id') or '?':<24} {_age}" + ) + # ========================================================================= # Deep checks # ========================================================================= diff --git a/tests/hermes_cli/test_cli_active_session_limit.py b/tests/hermes_cli/test_cli_active_session_limit.py index 6e47954c8a0ec..b62cef4593259 100644 --- a/tests/hermes_cli/test_cli_active_session_limit.py +++ b/tests/hermes_cli/test_cli_active_session_limit.py @@ -25,10 +25,10 @@ def test_cli_claim_active_session_respects_global_limit(tmp_path, monkeypatch): try: assert cli._claim_active_session("cli") is False - assert printed == [ - "[bold red]Hermes is at the active session limit (1/1). " - "Try again when another session finishes.[/]" - ] + assert len(printed) == 1 + assert "active session limit (1/1)" in printed[0] + # Names the holding surface ("tui"), not the blocked one. + assert "Held by: tui" in printed[0] held.release()