seb/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(sessions): add optional max session cap

Browse Source
This commit is contained in:
Robin Fernandes 2026-06-09 06:21:23 +10:00 committed by Teknium
parent 1e3b3dfabb
commit 639c1e3636
14 changed files with 1490 additions and 202 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
cli-config.yaml.example
View File

@ -528,6 +528,15 @@ session_reset:
idle_minutes: 1440 # Inactivity timeout in minutes (default: 1440 = 24 hours)
at_hour: 4 # Daily reset hour, 0-23 local time (default: 4 AM)
# Maximum number of simultaneously active chat sessions across CLI, TUI,
# dashboard chat, and messaging gateway. Set to null, 0, or omit to allow
# unlimited concurrent sessions. When the limit is reached, new sessions get a
# clean error while existing active sessions keep their normal behavior. This
# top-level key takes precedence over gateway.max_concurrent_sessions. The cap
# is a best-effort single-host/profile runtime guard; Hermes fails open if the
# local runtime lease registry cannot be read or locked.
max_concurrent_sessions: null
# When true, group/channel chats use one session per participant when the platform
# provides a user ID. This is the secure default and prevents users in the same
# room from sharing context, interrupts, and token costs. Set false only if you

451
cli.py
View File

@ -3462,6 +3462,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
self._image_counter = 0
self.preloaded_skills: list[str] = []
self._startup_skills_line_shown = False
self._active_session_lease = None
# Voice mode state (also reinitialized inside run() for interactive TUI).
self._voice_lock = threading.Lock()
@ -3490,6 +3491,45 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
self._background_tasks: Dict[str, threading.Thread] = {}
self._background_task_counter = 0
def _claim_active_session(self, surface: str = "cli", *, stderr: bool = False) -> bool:
"""Claim a global active-session slot for this CLI process."""
if self._active_session_lease is not None:
return True
try:
from hermes_cli.active_sessions import try_acquire_active_session
lease, message = try_acquire_active_session(
session_id=self.session_id,
surface=surface,
config=self.config,
)
except Exception as exc:
logger.warning("Failed to claim active session slot: %s", exc)
return True
if message:
if stderr:
print(message, file=sys.stderr)
else:
self._console_print(f"[bold red]{message}[/]")
return False
self._active_session_lease = lease
try:
atexit.register(self._release_active_session)
except Exception:
pass
return True
def _release_active_session(self) -> None:
lease = getattr(self, "_active_session_lease", None)
if lease is None:
return
try:
lease.release()
except Exception:
logger.debug("Failed to release active session slot", exc_info=True)
finally:
self._active_session_lease = None
def _invalidate(self, min_interval: float = 0.25) -> None:
"""Throttled UI repaint for high-frequency background updates.
@ -10497,6 +10537,9 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
def run(self):
"""Run the interactive CLI loop with persistent input at bottom."""
if not self._claim_active_session("cli"):
return
# Detect light/dark terminal mode now (before pt grabs the tty).
# Caches the result so subsequent _hex_to_ansi / style calls
# don't risk re-querying mid-render.
@ -12918,6 +12961,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
pass
_run_cleanup()
self._print_exit_summary()
self._release_active_session()
# Deferred relaunch: /update sets _pending_relaunch so the exec
# happens here — after prompt_toolkit has exited and fully restored
@ -13281,219 +13325,224 @@ def main(
# Handle single query mode
if query or image:
query, single_query_images = _collect_query_images(query, image)
# Kanban workers spawn with ``hermes chat -q "work kanban task <id>"``;
# the actual task description lives in the task body. Mirror the
# gateway/CLI behaviour for inbound images by scanning the body for
# local image paths and http(s) image URLs and attaching them to the
# worker's first turn. Without this, users who paste a screenshot
# path or URL into a kanban task body never get it routed to the
# model's vision input.
single_query_image_urls: list[str] = []
_kanban_task_id = os.environ.get("HERMES_KANBAN_TASK", "").strip()
if _kanban_task_id:
try:
from hermes_cli import kanban_db as _kb
from agent.image_routing import extract_image_refs as _extract_refs
_conn = _kb.connect()
if not cli._claim_active_session("cli", stderr=bool(quiet)):
sys.exit(1)
try:
query, single_query_images = _collect_query_images(query, image)
# Kanban workers spawn with ``hermes chat -q "work kanban task <id>"``;
# the actual task description lives in the task body. Mirror the
# gateway/CLI behaviour for inbound images by scanning the body for
# local image paths and http(s) image URLs and attaching them to the
# worker's first turn. Without this, users who paste a screenshot
# path or URL into a kanban task body never get it routed to the
# model's vision input.
single_query_image_urls: list[str] = []
_kanban_task_id = os.environ.get("HERMES_KANBAN_TASK", "").strip()
if _kanban_task_id:
try:
_task = _kb.get_task(_conn, _kanban_task_id)
finally:
try:
_conn.close()
except Exception:
pass
_body = getattr(_task, "body", "") if _task is not None else ""
if _body:
_kb_paths, _kb_urls = _extract_refs(_body)
if _kb_paths:
# Dedupe against any --image the user already passed.
_seen = {str(p) for p in single_query_images}
for _p in _kb_paths:
if _p not in _seen:
_seen.add(_p)
single_query_images.append(Path(_p))
if _kb_urls:
single_query_image_urls.extend(_kb_urls)
except Exception as _exc:
# Best-effort enrichment; never block worker startup on it.
logger.debug("kanban image-ref extraction failed: %s", _exc)
if quiet:
# Quiet mode: suppress banner, spinner, tool previews.
# Only print the final response and parseable session info.
cli.tool_progress_mode = "off"
if cli._ensure_runtime_credentials():
effective_query: Any = query
if single_query_images or single_query_image_urls:
# Honour the same image-routing decision used by the
# interactive path. With a vision-capable model (incl.
# custom-provider models declared via
# `model.supports_vision: true`), attach images natively
# as image_url content parts. Otherwise fall back to the
# text-pipeline (vision_analyze pre-description).
_img_mode = "text"
_build_parts = None
try:
from agent.image_routing import (
build_native_content_parts as _build_parts, # noqa: F811
)
from agent.image_routing import decide_image_input_mode
from hermes_cli.config import load_config
from hermes_cli import kanban_db as _kb
from agent.image_routing import extract_image_refs as _extract_refs
_img_mode = decide_image_input_mode(
(cli.provider or "").strip(),
(cli.model or "").strip(),
load_config(),
)
except Exception:
_img_mode = "text"
if _img_mode == "native" and _build_parts is not None:
_conn = _kb.connect()
try:
_task = _kb.get_task(_conn, _kanban_task_id)
finally:
try:
_parts, _skipped = _build_parts(
query if isinstance(query, str) else "",
[str(p) for p in single_query_images],
image_urls=list(single_query_image_urls) or None,
_conn.close()
except Exception:
pass
_body = getattr(_task, "body", "") if _task is not None else ""
if _body:
_kb_paths, _kb_urls = _extract_refs(_body)
if _kb_paths:
# Dedupe against any --image the user already passed.
_seen = {str(p) for p in single_query_images}
for _p in _kb_paths:
if _p not in _seen:
_seen.add(_p)
single_query_images.append(Path(_p))
if _kb_urls:
single_query_image_urls.extend(_kb_urls)
except Exception as _exc:
# Best-effort enrichment; never block worker startup on it.
logger.debug("kanban image-ref extraction failed: %s", _exc)
if quiet:
# Quiet mode: suppress banner, spinner, tool previews.
# Only print the final response and parseable session info.
cli.tool_progress_mode = "off"
if cli._ensure_runtime_credentials():
effective_query: Any = query
if single_query_images or single_query_image_urls:
# Honour the same image-routing decision used by the
# interactive path. With a vision-capable model (incl.
# custom-provider models declared via
# `model.supports_vision: true`), attach images natively
# as image_url content parts. Otherwise fall back to the
# text-pipeline (vision_analyze pre-description).
_img_mode = "text"
_build_parts = None
try:
from agent.image_routing import (
build_native_content_parts as _build_parts, # noqa: F811
)
if any(p.get("type") == "image_url" for p in _parts):
effective_query = _parts
else:
# All images unreadable — text fallback.
# ``_preprocess_images_with_vision`` only knows
# about local files; URLs would be lost there,
# so keep the original query text intact when
# only URLs were supplied.
from agent.image_routing import decide_image_input_mode
from hermes_cli.config import load_config
_img_mode = decide_image_input_mode(
(cli.provider or "").strip(),
(cli.model or "").strip(),
load_config(),
)
except Exception:
_img_mode = "text"
if _img_mode == "native" and _build_parts is not None:
try:
_parts, _skipped = _build_parts(
query if isinstance(query, str) else "",
[str(p) for p in single_query_images],
image_urls=list(single_query_image_urls) or None,
)
if any(p.get("type") == "image_url" for p in _parts):
effective_query = _parts
else:
# All images unreadable — text fallback.
# ``_preprocess_images_with_vision`` only knows
# about local files; URLs would be lost there,
# so keep the original query text intact when
# only URLs were supplied.
if single_query_images:
effective_query = cli._preprocess_images_with_vision(
query, single_query_images, announce=False,
)
except Exception:
if single_query_images:
effective_query = cli._preprocess_images_with_vision(
query, single_query_images, announce=False,
)
except Exception:
if single_query_images:
effective_query = cli._preprocess_images_with_vision(
query, single_query_images, announce=False,
)
elif single_query_images:
effective_query = cli._preprocess_images_with_vision(
query,
single_query_images,
announce=False,
)
turn_route = cli._resolve_turn_agent_config(effective_query)
if turn_route["signature"] != cli._active_agent_route_signature:
cli.agent = None
if cli._init_agent(
model_override=turn_route["model"],
runtime_override=turn_route["runtime"],
request_overrides=turn_route.get("request_overrides"),
):
cli.agent.quiet_mode = True
cli.agent.suppress_status_output = True
# Suppress streaming display callbacks so stdout stays
# machine-readable (no styled "Hermes" box, no tool-gen
# status lines). The response is printed once below.
cli.agent.stream_delta_callback = None
cli.agent.tool_gen_callback = None
try:
result = cli.agent.run_conversation(
user_message=effective_query,
conversation_history=cli.conversation_history,
)
except KeyboardInterrupt:
_emit_interrupted_session_end(cli, reason="keyboard_interrupt")
print(f"\nsession_id: {cli.session_id}", file=sys.stderr)
sys.exit(130)
# Sync session_id if mid-run compression created a
# continuation session. The exit line below reports
# session_id to stderr for automation wrappers; without
# this sync it would point at the ended parent.
if (
getattr(cli.agent, "session_id", None)
and cli.agent.session_id != cli.session_id
elif single_query_images:
effective_query = cli._preprocess_images_with_vision(
query,
single_query_images,
announce=False,
)
turn_route = cli._resolve_turn_agent_config(effective_query)
if turn_route["signature"] != cli._active_agent_route_signature:
cli.agent = None
if cli._init_agent(
model_override=turn_route["model"],
runtime_override=turn_route["runtime"],
request_overrides=turn_route.get("request_overrides"),
):
cli.session_id = cli.agent.session_id
response = result.get("final_response", "") if isinstance(result, dict) else str(result)
# Surface backend errors that produced no visible output
# (e.g. invalid model slug → provider 4xx). Mirrors the
# interactive CLI path. Write to stderr so piped stdout
# stays clean for automation wrappers.
if (
not response
and isinstance(result, dict)
and result.get("error")
and (result.get("failed") or result.get("partial"))
):
print(f"Error: {result['error']}", file=sys.stderr)
elif response:
print(response)
# Kanban goal-loop mode: a worker spawned for a
# goal_mode card keeps working in THIS session until an
# auxiliary judge agrees the card is done, the worker
# terminates the task itself, or the turn budget runs
# out (→ sticky block). Gated on the env vars the
# dispatcher sets in `_default_spawn`; a no-op for every
# normal worker and every non-kanban `-q` run.
if os.environ.get("HERMES_KANBAN_GOAL_MODE") == "1":
cli.agent.quiet_mode = True
cli.agent.suppress_status_output = True
# Suppress streaming display callbacks so stdout stays
# machine-readable (no styled "Hermes" box, no tool-gen
# status lines). The response is printed once below.
cli.agent.stream_delta_callback = None
cli.agent.tool_gen_callback = None
try:
_run_kanban_goal_loop_q(cli, response)
except Exception as _goal_exc:
logger.debug("kanban goal loop failed: %s", _goal_exc)
result = cli.agent.run_conversation(
user_message=effective_query,
conversation_history=cli.conversation_history,
)
except KeyboardInterrupt:
_emit_interrupted_session_end(cli, reason="keyboard_interrupt")
print(f"\nsession_id: {cli.session_id}", file=sys.stderr)
sys.exit(130)
# Sync session_id if mid-run compression created a
# continuation session. The exit line below reports
# session_id to stderr for automation wrappers; without
# this sync it would point at the ended parent.
if (
getattr(cli.agent, "session_id", None)
and cli.agent.session_id != cli.session_id
):
cli.session_id = cli.agent.session_id
response = result.get("final_response", "") if isinstance(result, dict) else str(result)
# Surface backend errors that produced no visible output
# (e.g. invalid model slug → provider 4xx). Mirrors the
# interactive CLI path. Write to stderr so piped stdout
# stays clean for automation wrappers.
if (
not response
and isinstance(result, dict)
and result.get("error")
and (result.get("failed") or result.get("partial"))
):
print(f"Error: {result['error']}", file=sys.stderr)
elif response:
print(response)
# Session ID goes to stderr so piped stdout is clean.
print(f"\nsession_id: {cli.session_id}", file=sys.stderr)
# Ensure proper exit code for automation wrappers.
#
# Kanban workers get a special case: when the run failed
# purely because the provider rate-limited / exhausted
# quota (not because the task itself is broken), exit with
# the EX_TEMPFAIL sentinel instead of the generic 1. The
# dispatcher's reap classifier maps that code to a
# ``rate_limited`` exit and releases the task back to
# ``ready`` WITHOUT incrementing the failure counter, so a
# 5-hour quota window can't trip the circuit breaker and
# permanently block the card. Non-kanban runs keep the
# plain 0/1 contract automation wrappers expect.
_exit_code = 0
if isinstance(result, dict) and result.get("failed"):
_exit_code = 1
if os.environ.get("HERMES_KANBAN_TASK") and result.get(
"failure_reason"
) in ("rate_limit", "billing"):
# Kanban goal-loop mode: a worker spawned for a
# goal_mode card keeps working in THIS session until an
# auxiliary judge agrees the card is done, the worker
# terminates the task itself, or the turn budget runs
# out (→ sticky block). Gated on the env vars the
# dispatcher sets in `_default_spawn`; a no-op for every
# normal worker and every non-kanban `-q` run.
if os.environ.get("HERMES_KANBAN_GOAL_MODE") == "1":
try:
from hermes_cli.kanban_db import (
KANBAN_RATE_LIMIT_EXIT_CODE as _RL_CODE,
)
_exit_code = _RL_CODE
except Exception:
_exit_code = 1
sys.exit(_exit_code)
# Exit with error code if credentials or agent init fails
sys.exit(1)
else:
# Single-query mode (`hermes chat -q "…"`): skip the welcome
# banner. Building the banner takes ~420 ms on cold start —
# ~200 ms of that is the version-update check, the rest is
# toolset / skill enumeration and Rich panel rendering. None
# of that is useful for a one-shot query: the user already
# picked the prompt, doesn't need a toolset reference, and
# gets the session ID + resume hint from
# ``_print_exit_summary()`` after the response prints.
#
# The fully-quiet ``-Q`` / ``--quiet`` machine-readable path
# above was already banner-free; this brings the human-
# facing single-query path in line so all non-interactive
# invocations are fast.
_query_label = query or ("[image attached]" if single_query_images else "")
if _query_label:
cli.console.print(f"[bold blue]Query:[/] {_query_label}")
# Surface security advisories before the agent runs — short
# banner, doesn't depend on the welcome banner being shown.
cli._show_security_advisories()
cli.chat(query, images=single_query_images or None)
cli._print_exit_summary()
_run_kanban_goal_loop_q(cli, response)
except Exception as _goal_exc:
logger.debug("kanban goal loop failed: %s", _goal_exc)
# Session ID goes to stderr so piped stdout is clean.
print(f"\nsession_id: {cli.session_id}", file=sys.stderr)
# Ensure proper exit code for automation wrappers.
#
# Kanban workers get a special case: when the run failed
# purely because the provider rate-limited / exhausted
# quota (not because the task itself is broken), exit with
# the EX_TEMPFAIL sentinel instead of the generic 1. The
# dispatcher's reap classifier maps that code to a
# ``rate_limited`` exit and releases the task back to
# ``ready`` WITHOUT incrementing the failure counter, so a
# 5-hour quota window can't trip the circuit breaker and
# permanently block the card. Non-kanban runs keep the
# plain 0/1 contract automation wrappers expect.
_exit_code = 0
if isinstance(result, dict) and result.get("failed"):
_exit_code = 1
if os.environ.get("HERMES_KANBAN_TASK") and result.get(
"failure_reason"
) in ("rate_limit", "billing"):
try:
from hermes_cli.kanban_db import (
KANBAN_RATE_LIMIT_EXIT_CODE as _RL_CODE,
)
_exit_code = _RL_CODE
except Exception:
_exit_code = 1
sys.exit(_exit_code)
# Exit with error code if credentials or agent init fails
sys.exit(1)
else:
# Single-query mode (`hermes chat -q "…"`): skip the welcome
# banner. Building the banner takes ~420 ms on cold start —
# ~200 ms of that is the version-update check, the rest is
# toolset / skill enumeration and Rich panel rendering. None
# of that is useful for a one-shot query: the user already
# picked the prompt, doesn't need a toolset reference, and
# gets the session ID + resume hint from
# ``_print_exit_summary()`` after the response prints.
#
# The fully-quiet ``-Q`` / ``--quiet`` machine-readable path
# above was already banner-free; this brings the human-
# facing single-query path in line so all non-interactive
# invocations are fast.
_query_label = query or ("[image attached]" if single_query_images else "")
if _query_label:
cli.console.print(f"[bold blue]Query:[/] {_query_label}")
# Surface security advisories before the agent runs — short
# banner, doesn't depend on the welcome banner being shown.
cli._show_security_advisories()
cli.chat(query, images=single_query_images or None)
cli._print_exit_summary()
finally:
cli._release_active_session()
return
# Run interactive mode

57
gateway/config.py
View File

@ -56,6 +56,42 @@ def _coerce_int(value: Any, default: int) -> int:
return default
def _coerce_optional_positive_int(value: Any, key: str) -> Optional[int]:
"""Coerce an optional positive integer config value.
``None``/0/negative disable the setting. Malformed values are ignored with
a warning so a typo never prevents the gateway from starting.
"""
if value is None:
return None
if isinstance(value, bool):
logger.warning(
"Ignoring invalid %s=%r (expected a positive integer; 0/null disables)",
key,
value,
)
return None
try:
if isinstance(value, float):
if not value.is_integer():
raise ValueError(value)
parsed = int(value)
elif isinstance(value, str):
parsed = int(value.strip(), 10)
else:
parsed = int(value)
except (TypeError, ValueError):
logger.warning(
"Ignoring invalid %s=%r (expected a positive integer; 0/null disables)",
key,
value,
)
return None
if parsed <= 0:
return None
return parsed
def _normalize_unauthorized_dm_behavior(value: Any, default: str = "pair") -> str:
"""Normalize unauthorized DM behavior to a supported value."""
if isinstance(value, str):
@ -495,6 +531,7 @@ class GatewayConfig:
# Session isolation in shared chats
group_sessions_per_user: bool = True # Isolate group/channel sessions per participant when user IDs are available
thread_sessions_per_user: bool = False # When False (default), threads are shared across all participants
max_concurrent_sessions: Optional[int] = None # Positive int caps simultaneous active chat sessions
# Unauthorized DM policy
unauthorized_dm_behavior: str = "pair" # "pair" or "ignore"
@ -600,6 +637,7 @@ class GatewayConfig:
"stt_enabled": self.stt_enabled,
"group_sessions_per_user": self.group_sessions_per_user,
"thread_sessions_per_user": self.thread_sessions_per_user,
"max_concurrent_sessions": self.max_concurrent_sessions,
"unauthorized_dm_behavior": self.unauthorized_dm_behavior,
"streaming": self.streaming.to_dict(),
"session_store_max_age_days": self.session_store_max_age_days,
@ -645,6 +683,17 @@ class GatewayConfig:
group_sessions_per_user = data.get("group_sessions_per_user")
thread_sessions_per_user = data.get("thread_sessions_per_user")
nested_gateway = data.get("gateway") if isinstance(data.get("gateway"), dict) else {}
if "max_concurrent_sessions" in data:
max_concurrent_raw = data.get("max_concurrent_sessions")
max_concurrent_key = "max_concurrent_sessions"
else:
max_concurrent_raw = nested_gateway.get("max_concurrent_sessions")
max_concurrent_key = "gateway.max_concurrent_sessions"
max_concurrent_sessions = _coerce_optional_positive_int(
max_concurrent_raw,
max_concurrent_key,
)
unauthorized_dm_behavior = _normalize_unauthorized_dm_behavior(
data.get("unauthorized_dm_behavior"),
"pair",
@ -671,6 +720,7 @@ class GatewayConfig:
stt_enabled=_coerce_bool(stt_enabled, True),
group_sessions_per_user=_coerce_bool(group_sessions_per_user, True),
thread_sessions_per_user=_coerce_bool(thread_sessions_per_user, False),
max_concurrent_sessions=max_concurrent_sessions,
unauthorized_dm_behavior=unauthorized_dm_behavior,
streaming=StreamingConfig.from_dict(data.get("streaming", {})),
session_store_max_age_days=session_store_max_age_days,
@ -761,6 +811,13 @@ def load_gateway_config() -> GatewayConfig:
if "thread_sessions_per_user" in yaml_cfg:
gw_data["thread_sessions_per_user"] = yaml_cfg["thread_sessions_per_user"]
gateway_section = yaml_cfg.get("gateway")
if isinstance(gateway_section, dict) and "max_concurrent_sessions" in gateway_section:
gw_data["max_concurrent_sessions"] = gateway_section["max_concurrent_sessions"]
if "max_concurrent_sessions" in yaml_cfg:
gw_data["max_concurrent_sessions"] = yaml_cfg["max_concurrent_sessions"]
streaming_cfg = yaml_cfg.get("streaming")
if not isinstance(streaming_cfg, dict):
# Fall back to nested gateway.streaming written by

78
gateway/run.py
View File

@ -1934,6 +1934,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
# Key: session_key, Value: AIAgent instance
self._running_agents: Dict[str, Any] = {}
self._running_agents_ts: Dict[str, float] = {} # start timestamp per session
self._active_session_leases: Dict[str, Any] = {}
self._pending_messages: Dict[str, str] = {} # Queued messages during interrupt
# Last successfully-resolved (non-empty) model, keyed by session. Used
# as a fallback when a fresh config read transiently returns an empty
@ -3390,6 +3391,59 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
if agent is not _AGENT_PENDING_SENTINEL
}
def _get_max_concurrent_sessions(self) -> Optional[int]:
"""Return the configured active chat session cap, if enabled."""
try:
from hermes_cli.active_sessions import resolve_max_concurrent_sessions
return resolve_max_concurrent_sessions(getattr(self, "config", None))
except Exception:
return None
def _active_session_limit_message(self, session_key: str) -> Optional[str]:
"""Return a user-facing rejection when starting a new session exceeds the cap."""
max_sessions = self._get_max_concurrent_sessions()
if max_sessions is None:
return None
if session_key in getattr(self, "_running_agents", {}):
return None
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."
)
def _claim_active_session_slot(
self,
session_key: str,
source: SessionSource,
) -> tuple[Any, Optional[str]]:
"""Claim a cross-process active-session slot for a new gateway turn."""
if session_key in getattr(self, "_running_agents", {}):
return None, None
local_limit_message = self._active_session_limit_message(session_key)
if local_limit_message is not None:
return None, local_limit_message
try:
from hermes_cli.active_sessions import try_acquire_active_session
platform = source.platform.value if source and source.platform else "gateway"
return try_acquire_active_session(
session_id=session_key,
surface=f"gateway:{platform}",
config=getattr(self, "config", None),
metadata={
"platform": platform,
"chat_id": getattr(source, "chat_id", "") or "",
"user_id": getattr(source, "user_id", "") or "",
},
)
except Exception as exc:
logger.warning("Failed to claim active session slot: %s", exc)
return None, None
@staticmethod
def _agent_has_active_subagents(running_agent: Any) -> bool:
"""Return True when *running_agent* is currently driving subagents
@ -5751,8 +5805,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
self._background_tasks.clear()
self.adapters.clear()
for _session_key in list(self._running_agents):
self._release_running_agent_state(_session_key)
self._running_agents.clear()
self._running_agents_ts.clear()
if hasattr(self, "_active_session_leases"):
self._active_session_leases.clear()
self._pending_messages.clear()
self._pending_approvals.clear()
if hasattr(self, '_busy_ack_ts'):
@ -7237,6 +7295,20 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
# message arriving during any of those yields would pass the
# "already running" guard and spin up a duplicate agent for the
# same session — corrupting the transcript.
_active_session_lease, _limit_message = self._claim_active_session_slot(
_quick_key,
source,
)
if _limit_message is not None:
logger.info(
"Rejecting new active session %s: max_concurrent_sessions reached",
_quick_key,
)
return _limit_message
if _active_session_lease is not None:
if not hasattr(self, "_active_session_leases"):
self._active_session_leases = {}
self._active_session_leases[_quick_key] = _active_session_lease
self._running_agents[_quick_key] = _AGENT_PENDING_SENTINEL
self._running_agents_ts[_quick_key] = time.time()
_run_generation = self._begin_session_run_generation(_quick_key)
@ -11976,6 +12048,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
session_key, run_generation
):
return False
lease = getattr(self, "_active_session_leases", {}).pop(session_key, None)
if lease is not None:
try:
lease.release()
except Exception:
logger.debug("Failed to release active session slot", exc_info=True)
self._running_agents.pop(session_key, None)
self._running_agents_ts.pop(session_key, None)
if hasattr(self, "_busy_ack_ts"):

320
hermes_cli/active_sessions.py Normal file
View File

@ -0,0 +1,320 @@
"""Cross-process active chat session leases.
The session database records persisted conversations. This module records
currently open chat surfaces, including idle CLI/TUI sessions that have not
written a transcript row yet.
"""
from __future__ import annotations
import json
import logging
import os
import time
import uuid
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Optional
from hermes_constants import get_hermes_home
logger = logging.getLogger(__name__)
def coerce_max_concurrent_sessions(value: Any, key: str = "max_concurrent_sessions") -> Optional[int]:
"""Return a positive integer cap, or None when disabled/invalid."""
if value is None:
return None
if isinstance(value, bool):
logger.warning(
"Ignoring invalid %s=%r (expected a positive integer; 0/null disables)",
key,
value,
)
return None
try:
if isinstance(value, float):
if not value.is_integer():
raise ValueError(value)
parsed = int(value)
elif isinstance(value, str):
parsed = int(value.strip(), 10)
else:
parsed = int(value)
except (TypeError, ValueError):
logger.warning(
"Ignoring invalid %s=%r (expected a positive integer; 0/null disables)",
key,
value,
)
return None
if parsed <= 0:
return None
return parsed
def resolve_max_concurrent_sessions(config: Any) -> Optional[int]:
"""Resolve top-level max_concurrent_sessions with gateway.* fallback."""
raw: Any = None
key = "max_concurrent_sessions"
if isinstance(config, dict):
if "max_concurrent_sessions" in config:
raw = config.get("max_concurrent_sessions")
else:
gateway_cfg = config.get("gateway")
if isinstance(gateway_cfg, dict):
raw = gateway_cfg.get("max_concurrent_sessions")
key = "gateway.max_concurrent_sessions"
else:
raw = getattr(config, "max_concurrent_sessions", None)
return coerce_max_concurrent_sessions(raw, key=key)
def active_session_limit_message(active_count: int, max_sessions: int) -> str:
return (
f"Hermes is at the active session limit ({active_count}/{max_sessions}). "
"Try again when another session finishes."
)
def _state_dir() -> Path:
return get_hermes_home() / "runtime"
def _state_path() -> Path:
return _state_dir() / "active_sessions.json"
def _lock_path() -> Path:
return _state_dir() / "active_sessions.lock"
class _FileLock:
def __init__(self, path: Path):
self.path = path
self._fh = None
def __enter__(self):
self.path.parent.mkdir(parents=True, exist_ok=True)
self._fh = open(self.path, "a+b")
if os.name == "nt":
try:
import msvcrt
self._fh.seek(0)
msvcrt.locking(self._fh.fileno(), msvcrt.LK_LOCK, 1)
except Exception as exc:
self._fh.close()
self._fh = None
raise RuntimeError("active session file lock unavailable") from exc
else:
try:
import fcntl
fcntl.flock(self._fh.fileno(), fcntl.LOCK_EX)
except Exception as exc:
self._fh.close()
self._fh = None
raise RuntimeError("active session file lock unavailable") from exc
return self
def __exit__(self, exc_type, exc, tb):
if self._fh is None:
return
if os.name == "nt":
try:
import msvcrt
self._fh.seek(0)
msvcrt.locking(self._fh.fileno(), msvcrt.LK_UNLCK, 1)
except Exception:
pass
else:
try:
import fcntl
fcntl.flock(self._fh.fileno(), fcntl.LOCK_UN)
except Exception:
pass
try:
self._fh.close()
finally:
self._fh = None
def _read_entries(path: Path) -> list[dict[str, Any]]:
try:
with open(path, "r", encoding="utf-8") as fh:
data = json.load(fh)
except FileNotFoundError:
return []
except Exception:
logger.warning("Ignoring corrupt active session registry at %s", path)
return []
entries = data.get("entries") if isinstance(data, dict) else data
if not isinstance(entries, list):
return []
return [entry for entry in entries if isinstance(entry, dict)]
def _write_entries(path: Path, entries: list[dict[str, Any]]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
tmp = path.with_name(f"{path.name}.{os.getpid()}.{uuid.uuid4().hex}.tmp")
with open(tmp, "w", encoding="utf-8") as fh:
json.dump({"entries": entries}, fh, sort_keys=True)
os.replace(tmp, path)
def _process_start_time(pid: int) -> Optional[float]:
# Pair pid with process create_time when psutil can read it, so a recycled
# pid does not keep a stale lease alive indefinitely.
try:
import psutil # type: ignore
return float(psutil.Process(pid).create_time())
except Exception:
return None
def _optional_float(value: Any) -> Optional[float]:
if value is None or value == "":
return None
try:
return float(value)
except (TypeError, ValueError):
return None
def _pid_alive(pid: Any, process_start_time: Any = None) -> bool:
try:
pid_int = int(pid)
except (TypeError, ValueError):
return False
if pid_int <= 0:
return False
try:
from gateway.status import _pid_exists
exists = bool(_pid_exists(pid_int))
except Exception:
return False
if not exists:
return False
expected_start = _optional_float(process_start_time)
if expected_start is None:
return True
current_start = _process_start_time(pid_int)
if current_start is None:
return True
return abs(current_start - expected_start) < 0.001
def _prune_dead(entries: list[dict[str, Any]]) -> list[dict[str, Any]]:
return [
entry
for entry in entries
if _pid_alive(entry.get("pid"), entry.get("process_start_time"))
]
@dataclass
class ActiveSessionLease:
lease_id: str
session_id: str
surface: str
enabled: bool = True
released: bool = False
def release(self) -> None:
if self.released or not self.enabled:
return
release_active_session(self)
def try_acquire_active_session(
*,
session_id: str,
surface: str,
config: Any,
metadata: Optional[dict[str, Any]] = None,
) -> tuple[Optional[ActiveSessionLease], Optional[str]]:
"""Acquire an active-session slot.
Returns ``(lease, None)`` on success. When the cap is disabled, the lease is
a no-op object so callers can unconditionally call ``release()``.
"""
max_sessions = resolve_max_concurrent_sessions(config)
lease_id = uuid.uuid4().hex
if max_sessions is None:
return ActiveSessionLease(
lease_id=lease_id,
session_id=session_id,
surface=surface,
enabled=False,
), None
now = time.time()
entry = {
"lease_id": lease_id,
"session_id": str(session_id),
"surface": str(surface),
"pid": os.getpid(),
"process_start_time": _process_start_time(os.getpid()),
"started_at": now,
"updated_at": now,
}
if metadata:
entry["metadata"] = {
str(k): v for k, v in metadata.items() if isinstance(k, str)
}
state_path = _state_path()
with _FileLock(_lock_path()):
raw_entries = _read_entries(state_path)
entries = _prune_dead(raw_entries)
pruned = len(raw_entries) - len(entries)
if pruned:
logger.info("Pruned %d stale active session lease(s)", pruned)
active_count = len(entries)
if active_count >= max_sessions:
_write_entries(state_path, entries)
logger.info(
"Active session limit reached: active=%d max=%d surface=%s",
active_count,
max_sessions,
surface,
)
return None, active_session_limit_message(active_count, max_sessions)
entries.append(entry)
_write_entries(state_path, entries)
return ActiveSessionLease(
lease_id=lease_id,
session_id=str(session_id),
surface=str(surface),
), None
def release_active_session(lease: ActiveSessionLease) -> None:
state_path = _state_path()
try:
with _FileLock(_lock_path()):
entries = _prune_dead(_read_entries(state_path))
kept = [
entry
for entry in entries
if str(entry.get("lease_id") or "") != lease.lease_id
]
if len(kept) != len(entries):
_write_entries(state_path, kept)
finally:
lease.released = True
def active_session_registry_snapshot() -> list[dict[str, Any]]:
"""Return the pruned active-session registry for diagnostics/tests."""
state_path = _state_path()
with _FileLock(_lock_path()):
entries = _prune_dead(_read_entries(state_path))
_write_entries(state_path, entries)
return entries

3
hermes_cli/config.py
View File

@ -805,6 +805,9 @@ DEFAULT_CONFIG = {
"fallback_providers": [],
"credential_pool_strategies": {},
"toolsets": ["hermes-cli"],
# Global active chat session cap across CLI, TUI/dashboard, and messaging.
# None/0 = unbounded.
"max_concurrent_sessions": None,
"agent": {
"max_turns": 90,
# Inactivity timeout for gateway agent execution (seconds).

83
tests/gateway/test_config.py
View File

@ -1,5 +1,6 @@
"""Tests for gateway configuration management."""
import logging
import os
from unittest.mock import patch
@ -213,6 +214,43 @@ class TestGatewayConfigRoundtrip:
assert restored.group_sessions_per_user is False
assert restored.thread_sessions_per_user is True
def test_max_concurrent_sessions_from_dict_normalizes_disabled_values(self):
assert GatewayConfig.from_dict({}).max_concurrent_sessions is None
assert GatewayConfig.from_dict({"max_concurrent_sessions": None}).max_concurrent_sessions is None
assert GatewayConfig.from_dict({"max_concurrent_sessions": 0}).max_concurrent_sessions is None
assert GatewayConfig.from_dict({"max_concurrent_sessions": -1}).max_concurrent_sessions is None
def test_max_concurrent_sessions_from_dict_accepts_positive_integer(self):
config = GatewayConfig.from_dict({"max_concurrent_sessions": "3"})
assert config.max_concurrent_sessions == 3
def test_max_concurrent_sessions_from_dict_ignores_invalid_values(self, caplog):
caplog.set_level(logging.WARNING, logger="gateway.config")
config = GatewayConfig.from_dict({"max_concurrent_sessions": "many"})
assert config.max_concurrent_sessions is None
assert any(
"Ignoring invalid max_concurrent_sessions='many'" in record.message
for record in caplog.records
)
def test_max_concurrent_sessions_from_dict_accepts_nested_fallback(self):
config = GatewayConfig.from_dict({"gateway": {"max_concurrent_sessions": 4}})
assert config.max_concurrent_sessions == 4
def test_max_concurrent_sessions_top_level_overrides_nested(self):
config = GatewayConfig.from_dict(
{
"gateway": {"max_concurrent_sessions": 4},
"max_concurrent_sessions": 2,
}
)
assert config.max_concurrent_sessions == 2
def test_roundtrip_preserves_unauthorized_dm_behavior(self):
config = GatewayConfig(
unauthorized_dm_behavior="ignore",
@ -309,6 +347,51 @@ class TestLoadGatewayConfig:
assert config.thread_sessions_per_user is False
def test_bridges_top_level_max_concurrent_sessions_from_config_yaml(self, tmp_path, monkeypatch):
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
config_path = hermes_home / "config.yaml"
config_path.write_text("max_concurrent_sessions: 2\n", encoding="utf-8")
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
config = load_gateway_config()
assert config.max_concurrent_sessions == 2
def test_bridges_nested_max_concurrent_sessions_from_config_yaml(self, tmp_path, monkeypatch):
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
config_path = hermes_home / "config.yaml"
config_path.write_text(
"gateway:\n"
" max_concurrent_sessions: 3\n",
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
config = load_gateway_config()
assert config.max_concurrent_sessions == 3
def test_top_level_max_concurrent_sessions_overrides_nested_config_yaml(self, tmp_path, monkeypatch):
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
config_path = hermes_home / "config.yaml"
config_path.write_text(
"max_concurrent_sessions: 2\n"
"gateway:\n"
" max_concurrent_sessions: 3\n",
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
config = load_gateway_config()
assert config.max_concurrent_sessions == 2
def test_bridges_discord_thread_require_mention_from_config_yaml(self, tmp_path, monkeypatch):
"""discord.thread_require_mention in config.yaml should reach the runtime env var."""
hermes_home = tmp_path / ".hermes"

208
tests/gateway/test_max_concurrent_sessions.py Normal file
View File

@ -0,0 +1,208 @@
"""Tests for the gateway max_concurrent_sessions active-session cap."""
import asyncio
import time
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from gateway.config import GatewayConfig, Platform, PlatformConfig
from gateway.platforms.base import MessageEvent, MessageType
from gateway.run import GatewayRunner, _AGENT_PENDING_SENTINEL
from gateway.session import SessionSource, build_session_key
@pytest.fixture(autouse=True)
def _isolated_active_session_registry(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
class _FakeAdapter:
def __init__(self):
self._pending_messages = {}
self._active_sessions = {}
async def send(self, chat_id, text, **kwargs):
return None
async def interrupt_session_activity(self, session_key, chat_id):
event = self._active_sessions.get(session_key)
if event is not None:
event.set()
def _make_source(chat_id: str = "chat-1") -> SessionSource:
return SessionSource(
platform=Platform.TELEGRAM,
chat_id=chat_id,
chat_type="dm",
user_id=f"user-{chat_id}",
)
def _make_event(text: str = "hello", chat_id: str = "chat-1") -> MessageEvent:
return MessageEvent(
text=text,
message_type=MessageType.TEXT,
source=_make_source(chat_id),
)
def _make_runner(max_concurrent_sessions: int | None = None) -> GatewayRunner:
runner = object.__new__(GatewayRunner)
runner.config = GatewayConfig(
platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="***")},
max_concurrent_sessions=max_concurrent_sessions,
)
runner.adapters = {Platform.TELEGRAM: _FakeAdapter()}
runner._running_agents = {}
runner._running_agents_ts = {}
runner._active_session_leases = {}
runner._session_run_generation = {}
runner._pending_messages = {}
runner._pending_approvals = {}
runner._voice_mode = {}
runner._background_tasks = set()
runner._draining = False
runner._restart_requested = False
runner._restart_task_started = False
runner._restart_detached = False
runner._restart_via_service = False
runner._restart_drain_timeout = 0.0
runner._stop_task = None
runner._exit_code = None
runner._busy_ack_ts = {}
runner._busy_input_mode = "interrupt"
runner._busy_text_mode = "interrupt"
runner._queued_events = {}
runner._update_runtime_status = MagicMock()
runner._is_user_authorized = lambda _source: True
runner.hooks = MagicMock()
runner.hooks.emit = AsyncMock()
runner.session_store = MagicMock()
runner.delivery_router = MagicMock()
return runner
def _occupy_session(runner: GatewayRunner, chat_id: str = "busy"):
source = _make_source(chat_id)
session_key = build_session_key(source)
runner._running_agents[session_key] = MagicMock()
runner._running_agents_ts[session_key] = time.time()
return session_key
def _silence_global_gateway_hooks(monkeypatch):
monkeypatch.setattr("hermes_cli.plugins.invoke_hook", lambda *args, **kwargs: [])
monkeypatch.setattr("tools.slash_confirm.get_pending", lambda *args, **kwargs: None)
monkeypatch.setattr("tools.slash_confirm.clear_if_stale", lambda *args, **kwargs: None)
monkeypatch.setattr("tools.approval.has_blocking_approval", lambda *args, **kwargs: False)
def test_new_session_gets_clean_error_at_active_session_limit(monkeypatch):
_silence_global_gateway_hooks(monkeypatch)
runner = _make_runner(max_concurrent_sessions=1)
_occupy_session(runner, "busy")
event = _make_event(chat_id="new")
new_key = build_session_key(event.source)
async def fail_if_agent_runs(self_inner, ev, src, qk, generation):
raise AssertionError("_handle_message_with_agent should not run at capacity")
with patch.object(GatewayRunner, "_handle_message_with_agent", fail_if_agent_runs):
result = asyncio.run(runner._handle_message(event))
assert result == (
"Hermes is at the active session limit (1/1). "
"Try again when another session finishes."
)
assert new_key not in runner._running_agents
runner.session_store.get_or_create_session.assert_not_called()
def test_existing_active_session_uses_busy_handling_at_limit(monkeypatch):
_silence_global_gateway_hooks(monkeypatch)
runner = _make_runner(max_concurrent_sessions=1)
runner._busy_input_mode = "queue"
event = _make_event(chat_id="busy")
session_key = build_session_key(event.source)
runner._running_agents[session_key] = MagicMock()
runner._running_agents_ts[session_key] = 0
async def fail_if_agent_runs(self_inner, ev, src, qk, generation):
raise AssertionError("_handle_message_with_agent should not run for busy follow-up")
with patch.object(GatewayRunner, "_handle_message_with_agent", fail_if_agent_runs):
result = asyncio.run(runner._handle_message(event))
assert result is None
assert runner.adapters[Platform.TELEGRAM]._pending_messages[session_key] is event
def test_new_session_can_start_after_active_session_released(monkeypatch):
_silence_global_gateway_hooks(monkeypatch)
runner = _make_runner(max_concurrent_sessions=1)
busy_key = _occupy_session(runner, "busy")
runner._release_running_agent_state(busy_key)
event = _make_event(chat_id="new")
sentinel_seen = False
async def mock_agent_run(self_inner, ev, src, qk, generation):
nonlocal sentinel_seen
sentinel_seen = runner._running_agents.get(qk) is _AGENT_PENDING_SENTINEL
return "ok"
with patch.object(GatewayRunner, "_handle_message_with_agent", mock_agent_run):
result = asyncio.run(runner._handle_message(event))
assert result == "ok"
assert sentinel_seen is True
def test_status_command_bypasses_active_session_limit(monkeypatch):
_silence_global_gateway_hooks(monkeypatch)
runner = _make_runner(max_concurrent_sessions=1)
_occupy_session(runner, "busy")
runner._handle_status_command = AsyncMock(return_value="status ok")
result = asyncio.run(runner._handle_message(_make_event("/status", chat_id="new")))
assert result == "status ok"
runner._handle_status_command.assert_awaited_once()
def test_skill_command_that_would_start_agent_is_blocked_at_limit(monkeypatch):
_silence_global_gateway_hooks(monkeypatch)
runner = _make_runner(max_concurrent_sessions=1)
_occupy_session(runner, "busy")
monkeypatch.setattr(
"agent.skill_commands.get_skill_commands",
lambda: {"demo": {"name": "demo-skill"}},
)
monkeypatch.setattr(
"agent.skill_commands.resolve_skill_command_key",
lambda command: "demo" if command == "demo" else None,
)
monkeypatch.setattr(
"agent.skill_commands.build_skill_invocation_message",
lambda *args, **kwargs: "invoke demo skill",
)
monkeypatch.setattr(
"agent.skill_utils.get_disabled_skill_names",
lambda *args, **kwargs: [],
)
async def fail_if_agent_runs(self_inner, ev, src, qk, generation):
raise AssertionError("_handle_message_with_agent should not run at capacity")
with patch.object(GatewayRunner, "_handle_message_with_agent", fail_if_agent_runs):
result = asyncio.run(
runner._handle_message(_make_event("/demo please", chat_id="new"))
)
assert result == (
"Hermes is at the active session limit (1/1). "
"Try again when another session finishes."
)

6
tests/gateway/test_shutdown_cache_cleanup.py
View File

@ -84,6 +84,12 @@ class _FakeGateway:
def _evict_cached_agent(self, key):
pass
def _release_running_agent_state(self, session_key, **_kwargs):
agent = self._running_agents.pop(session_key, None)
self._running_agents_ts.pop(session_key, None)
self._cleanup_agent_resources(agent)
return agent is not None
def _make_mock_agent():
a = MagicMock()

313
tests/hermes_cli/test_active_sessions.py Normal file
View File

@ -0,0 +1,313 @@
import logging
import os
import subprocess
import sys
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from hermes_cli import active_sessions
def test_resolve_max_concurrent_sessions_values(caplog):
assert active_sessions.resolve_max_concurrent_sessions({}) is None
assert active_sessions.resolve_max_concurrent_sessions({"max_concurrent_sessions": None}) is None
assert active_sessions.resolve_max_concurrent_sessions({"max_concurrent_sessions": 0}) is None
assert active_sessions.resolve_max_concurrent_sessions({"max_concurrent_sessions": -1}) is None
assert active_sessions.resolve_max_concurrent_sessions({"max_concurrent_sessions": "3"}) == 3
assert (
active_sessions.resolve_max_concurrent_sessions(
{"gateway": {"max_concurrent_sessions": 4}}
)
== 4
)
assert (
active_sessions.resolve_max_concurrent_sessions(
{"max_concurrent_sessions": 2, "gateway": {"max_concurrent_sessions": 4}}
)
== 2
)
caplog.set_level(logging.WARNING)
assert active_sessions.resolve_max_concurrent_sessions({"max_concurrent_sessions": "many"}) is None
assert any(
"Ignoring invalid max_concurrent_sessions='many'" in record.message
for record in caplog.records
)
def test_active_session_lease_blocks_until_release(tmp_path, monkeypatch):
home = tmp_path / ".hermes"
monkeypatch.setenv("HERMES_HOME", str(home))
cfg = {"max_concurrent_sessions": 1}
lease, message = active_sessions.try_acquire_active_session(
session_id="session-1",
surface="cli",
config=cfg,
)
assert message is None
assert lease is not None
blocked_lease, blocked_message = active_sessions.try_acquire_active_session(
session_id="session-2",
surface="tui",
config=cfg,
)
assert blocked_lease is None
assert blocked_message == (
"Hermes is at the active session limit (1/1). "
"Try again when another session finishes."
)
lease.release()
next_lease, next_message = active_sessions.try_acquire_active_session(
session_id="session-3",
surface="gateway:telegram",
config=cfg,
)
assert next_message is None
assert next_lease is not None
next_lease.release()
assert active_sessions.active_session_registry_snapshot() == []
def test_active_session_registry_prunes_dead_pids(tmp_path, monkeypatch):
home = tmp_path / ".hermes"
monkeypatch.setenv("HERMES_HOME", str(home))
monkeypatch.setattr(
"gateway.status._pid_exists",
lambda pid: int(pid) != 99999999,
)
runtime = home / "runtime"
runtime.mkdir(parents=True)
active_sessions._write_entries(
runtime / "active_sessions.json",
[
{
"lease_id": "stale",
"session_id": "stale-session",
"surface": "cli",
"pid": 99999999,
"started_at": 1,
"updated_at": 1,
}
],
)
lease, message = active_sessions.try_acquire_active_session(
session_id="session-1",
surface="cli",
config={"max_concurrent_sessions": 1},
)
assert message is None
assert lease is not None
assert [entry["session_id"] for entry in active_sessions.active_session_registry_snapshot()] == [
"session-1"
]
lease.release()
def test_pid_alive_uses_safe_pid_exists_without_signalling(monkeypatch):
checked: list[int] = []
monkeypatch.setattr(
active_sessions.os,
"kill",
lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("os.kill used")),
)
monkeypatch.setattr(
"gateway.status._pid_exists",
lambda pid: checked.append(int(pid)) or True,
)
assert active_sessions._pid_alive(12345) is True
assert checked == [12345]
def test_active_session_hard_exit_is_reclaimed(tmp_path, monkeypatch):
home = tmp_path / ".hermes"
monkeypatch.setenv("HERMES_HOME", str(home))
repo_root = Path(__file__).resolve().parents[2]
env = os.environ.copy()
env["HERMES_HOME"] = str(home)
env["PYTHONPATH"] = str(repo_root)
child = subprocess.run(
[
sys.executable,
"-c",
(
"import os\n"
"from hermes_cli.active_sessions import try_acquire_active_session\n"
"lease, message = try_acquire_active_session("
"session_id='crash-session', surface='cli', "
"config={'max_concurrent_sessions': 1})\n"
"assert message is None, message\n"
"print(os.getpid(), flush=True)\n"
"os._exit(0)\n"
),
],
env=env,
text=True,
capture_output=True,
timeout=10,
check=True,
)
child_pid = int(child.stdout.strip())
lease, message = active_sessions.try_acquire_active_session(
session_id="next-session",
surface="cli",
config={"max_concurrent_sessions": 1},
)
assert child_pid > 0
assert message is None
assert lease is not None
assert [entry["session_id"] for entry in active_sessions.active_session_registry_snapshot()] == [
"next-session"
]
lease.release()
def test_concurrent_acquire_claims_only_one_last_slot(tmp_path, monkeypatch):
home = tmp_path / ".hermes"
monkeypatch.setenv("HERMES_HOME", str(home))
cfg = {"max_concurrent_sessions": 1}
def _claim(index: int):
return active_sessions.try_acquire_active_session(
session_id=f"session-{index}",
surface="cli",
config=cfg,
)
with ThreadPoolExecutor(max_workers=8) as pool:
results = list(pool.map(_claim, range(8)))
leases = [lease for lease, message in results if lease is not None and message is None]
blocked = [message for lease, message in results if lease is None and message]
try:
assert len(leases) == 1
assert len(blocked) == 7
assert active_sessions.active_session_registry_snapshot()[0]["session_id"].startswith("session-")
finally:
for lease in leases:
lease.release()
def test_cross_process_acquire_claims_only_one_last_slot(tmp_path, monkeypatch):
home = tmp_path / ".hermes"
monkeypatch.setenv("HERMES_HOME", str(home))
repo_root = Path(__file__).resolve().parents[2]
ready_dir = tmp_path / "ready"
ready_dir.mkdir()
go_file = tmp_path / "go"
env = os.environ.copy()
env["HERMES_HOME"] = str(home)
env["PYTHONPATH"] = str(repo_root)
script = (
"import os, time\n"
"from pathlib import Path\n"
"from hermes_cli.active_sessions import try_acquire_active_session\n"
"idx = os.environ['WORKER_INDEX']\n"
"ready_dir = Path(os.environ['READY_DIR'])\n"
"go_file = Path(os.environ['GO_FILE'])\n"
"(ready_dir / idx).write_text('ready', encoding='utf-8')\n"
"deadline = time.time() + 10\n"
"while not go_file.exists():\n"
" if time.time() > deadline:\n"
" raise RuntimeError('timed out waiting for go file')\n"
" time.sleep(0.01)\n"
"lease, message = try_acquire_active_session(\n"
" session_id=f'process-{idx}',\n"
" surface='cli',\n"
" config={'max_concurrent_sessions': 1},\n"
")\n"
"if lease is None:\n"
" print('BLOCK', flush=True)\n"
"else:\n"
" print('OK', flush=True)\n"
" time.sleep(2.0)\n"
" lease.release()\n"
)
workers: list[subprocess.Popen[str]] = []
try:
for index in range(6):
worker_env = env.copy()
worker_env["WORKER_INDEX"] = str(index)
worker_env["READY_DIR"] = str(ready_dir)
worker_env["GO_FILE"] = str(go_file)
workers.append(
subprocess.Popen(
[sys.executable, "-c", script],
env=worker_env,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
)
deadline = time.time() + 10
while len(list(ready_dir.iterdir())) < len(workers):
if time.time() > deadline:
raise AssertionError("workers did not become ready")
time.sleep(0.01)
go_file.write_text("go", encoding="utf-8")
outputs = []
for worker in workers:
stdout, stderr = worker.communicate(timeout=10)
assert worker.returncode == 0, stderr
outputs.append(stdout.strip())
finally:
for worker in workers:
if worker.poll() is None:
worker.kill()
worker.communicate()
assert outputs.count("OK") == 1
assert outputs.count("BLOCK") == len(workers) - 1
assert active_sessions.active_session_registry_snapshot() == []
def test_pid_start_time_mismatch_prunes_reused_pid(tmp_path, monkeypatch):
home = tmp_path / ".hermes"
monkeypatch.setenv("HERMES_HOME", str(home))
monkeypatch.setattr("gateway.status._pid_exists", lambda _pid: True)
monkeypatch.setattr(active_sessions, "_process_start_time", lambda _pid: 200.0)
runtime = home / "runtime"
runtime.mkdir(parents=True)
active_sessions._write_entries(
runtime / "active_sessions.json",
[
{
"lease_id": "stale-reused-pid",
"session_id": "stale-session",
"surface": "cli",
"pid": os.getpid(),
"process_start_time": 100.0,
"started_at": 1,
"updated_at": 1,
}
],
)
lease, message = active_sessions.try_acquire_active_session(
session_id="new-session",
surface="cli",
config={"max_concurrent_sessions": 1},
)
assert message is None
assert lease is not None
assert [entry["session_id"] for entry in active_sessions.active_session_registry_snapshot()] == [
"new-session"
]
lease.release()

41
tests/hermes_cli/test_cli_active_session_limit.py Normal file
View File

@ -0,0 +1,41 @@
from cli import HermesCLI
from hermes_cli.active_sessions import (
active_session_registry_snapshot,
try_acquire_active_session,
)
def test_cli_claim_active_session_respects_global_limit(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
cfg = {"max_concurrent_sessions": 1}
held, message = try_acquire_active_session(
session_id="held-session",
surface="tui",
config=cfg,
)
assert message is None
assert held is not None
cli = object.__new__(HermesCLI)
cli.session_id = "new-cli-session"
cli.config = cfg
cli._active_session_lease = None
printed: list[str] = []
cli._console_print = lambda text: printed.append(text)
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.[/]"
]
held.release()
assert cli._claim_active_session("cli") is True
assert [entry["session_id"] for entry in active_session_registry_snapshot()] == [
"new-cli-session"
]
finally:
held.release()
cli._release_active_session()

46
tests/test_tui_gateway_server.py
View File

@ -9,9 +9,55 @@ from datetime import datetime
from pathlib import Path
from unittest.mock import patch
from hermes_constants import reset_hermes_home_override, set_hermes_home_override
from hermes_cli.active_sessions import active_session_registry_snapshot
from tui_gateway import server
def test_session_create_rejects_at_active_session_limit(monkeypatch, tmp_path):
home = tmp_path / ".hermes"
home.mkdir()
(home / "config.yaml").write_text("max_concurrent_sessions: 1\n", encoding="utf-8")
token = set_hermes_home_override(home)
def _clear_server_sessions():
for session in list(server._sessions.values()):
server._teardown_session(session)
server._sessions.clear()
try:
server._cfg_cache = None
server._cfg_mtime = None
server._cfg_path = None
_clear_server_sessions()
monkeypatch.setattr(server, "_start_agent_build", lambda *args, **kwargs: None)
monkeypatch.setattr(server, "_completion_cwd", lambda params=None: str(tmp_path))
first = server._methods["session.create"]("r1", {"cols": 80})
assert "result" in first
sid = first["result"]["session_id"]
second = server._methods["session.create"]("r2", {"cols": 80})
assert second["error"]["message"] == (
"Hermes is at the active session limit (1/1). "
"Try again when another session finishes."
)
assert list(server._sessions) == [sid]
closed = server._methods["session.close"]("r3", {"session_id": sid})
assert closed["result"]["closed"] is True
assert active_session_registry_snapshot() == []
third = server._methods["session.create"]("r4", {"cols": 80})
assert "result" in third
finally:
_clear_server_sessions()
server._cfg_cache = None
server._cfg_mtime = None
server._cfg_path = None
reset_hermes_home_override(token)
def test_session_context_uses_session_cwd(monkeypatch, tmp_path):
"""Desktop/TUI sessions must pin the agent cwd per session.

58
tui_gateway/server.py
View File

@ -345,11 +345,44 @@ def _notify_session_boundary(event_type: str, session_id: str | None) -> None:
pass
def _claim_active_session_slot(
session_key: str,
*,
live_session_id: str,
surface: str = "tui",
) -> tuple[Any, str | None]:
try:
from hermes_cli.active_sessions import try_acquire_active_session
return try_acquire_active_session(
session_id=session_key,
surface=surface,
config=_load_cfg(),
metadata={"live_session_id": live_session_id},
)
except Exception as exc:
logger.warning("Failed to claim active session slot: %s", exc)
return None, None
def _release_active_session_slot(session: dict | None) -> None:
if not session:
return
lease = session.pop("active_session_lease", None)
if lease is None:
return
try:
lease.release()
except Exception:
logger.debug("Failed to release active session slot", exc_info=True)
def _finalize_session(session: dict | None, end_reason: str = "tui_close") -> None:
"""Best-effort finalize hook + memory commit for a session."""
if not session or session.get("_finalized"):
return
session["_finalized"] = True
_release_active_session_slot(session)
stop_event = session.get("_notif_stop")
if stop_event is not None:
stop_event.set()
@ -3284,6 +3317,9 @@ def _(rid, params: dict) -> dict:
ready = threading.Event()
now = time.time()
lease, limit_message = _claim_active_session_slot(key, live_session_id=sid)
if limit_message is not None:
return _err(rid, 4090, limit_message)
with _sessions_lock:
_sessions[sid] = {
@ -3292,6 +3328,7 @@ def _(rid, params: dict) -> dict:
"agent_ready": ready,
"attached_images": [],
"close_on_disconnect": is_truthy_value(params.get("close_on_disconnect", False)),
"active_session_lease": lease,
"cols": cols,
"created_at": now,
"edit_snapshots": {},
@ -3497,6 +3534,9 @@ def _(rid, params: dict) -> dict:
# _session_resume_lock across it would stall session.close on the main
# dispatch thread (it's not a _LONG_HANDLER), blocking fast-path RPCs.
sid = uuid.uuid4().hex[:8]
lease, limit_message = _claim_active_session_slot(target, live_session_id=sid)
if limit_message is not None:
return _err(rid, 4090, limit_message)
_enable_gateway_prompts()
home_token = (
set_hermes_home_override(str(profile_home)) if profile_home is not None else None
@ -3520,6 +3560,8 @@ def _(rid, params: dict) -> dict:
finally:
_clear_session_context(tokens)
except Exception as e:
if lease is not None:
lease.release()
return _err(rid, 5000, f"resume failed: {e}")
finally:
if home_token is not None:
@ -3536,6 +3578,8 @@ def _(rid, params: dict) -> dict:
agent.close()
except Exception:
pass
if lease is not None:
lease.release()
other_sid, other_session = live
payload = _live_session_payload(
other_sid,
@ -3555,7 +3599,10 @@ def _(rid, params: dict) -> dict:
# skills — must resolve to the resumed profile too).
if profile_home is not None:
_sessions[sid]["profile_home"] = str(profile_home)
_sessions[sid]["active_session_lease"] = lease
except Exception as e:
if lease is not None:
lease.release()
return _err(rid, 5000, f"resume failed: {e}")
session = _sessions.get(sid) or {}
return _ok(
@ -4192,6 +4239,10 @@ def _(rid, params: dict) -> dict:
if not history:
return _err(rid, 4008, "nothing to branch — send a message first")
new_key = _new_session_key()
new_sid = uuid.uuid4().hex[:8]
lease, limit_message = _claim_active_session_slot(new_key, live_session_id=new_sid)
if limit_message is not None:
return _err(rid, 4090, limit_message)
branch_name = params.get("name", "")
try:
if branch_name:
@ -4224,8 +4275,9 @@ def _(rid, params: dict) -> dict:
)
db.set_session_title(new_key, title)
except Exception as e:
if lease is not None:
lease.release()
return _err(rid, 5008, f"branch failed: {e}")
new_sid = uuid.uuid4().hex[:8]
try:
tokens = _set_session_context(new_key)
try:
@ -4235,7 +4287,11 @@ def _(rid, params: dict) -> dict:
_init_session(
new_sid, new_key, agent, list(history), cols=session.get("cols", 80)
)
if new_sid in _sessions:
_sessions[new_sid]["active_session_lease"] = lease
except Exception as e:
if lease is not None:
lease.release()
return _err(rid, 5000, f"agent init failed on branch: {e}")
return _ok(rid, {"session_id": new_sid, "title": title, "parent": old_key})

19
website/docs/user-guide/configuration.md
View File

@ -1417,6 +1417,25 @@ The master `streaming.enabled` switch is `false` by default — nothing streams
## Group Chat Session Isolation
Limit how many chat sessions can actively be open across CLI, TUI/dashboard,
and messaging gateway:
```yaml
max_concurrent_sessions: null # null/0 = unlimited; positive integer = active session cap
```
When the cap is reached, Hermes returns a direct limit message for new sessions.
Existing active sessions keep their normal behavior.
The canonical key is top-level `max_concurrent_sessions`. Hermes also accepts
`gateway.max_concurrent_sessions` as a fallback, but the top-level key wins when
both are set.
The cap is enforced with a local runtime lease file and is best-effort: Hermes
fails open if the registry cannot be read or locked so users are not stranded.
It is intended for a single host/profile runtime, not a shared `$HERMES_HOME`
mounted across multiple machines.
Control whether shared chats keep one conversation per room or one conversation per participant:
```yaml