mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-10-20 05:15:31 +00:00
Merge pull request #73497 from NousResearch/bb/active-session-clarity
fix(sessions): session cap counted invisible sessions and never said who held them
This commit is contained in:
commit
e581d92429
8 changed files with 228 additions and 54 deletions
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -70,10 +70,47 @@ def resolve_max_concurrent_sessions(config: Any) -> Optional[int]:
|
|||
return coerce_max_concurrent_sessions(raw, key=key)
|
||||
|
||||
|
||||
def active_session_limit_message(active_count: int, max_sessions: int) -> str:
|
||||
def format_age(seconds: float) -> str:
|
||||
minutes = max(0, int(seconds // 60))
|
||||
if minutes < 60:
|
||||
return f"{minutes}m"
|
||||
hours, minutes = divmod(minutes, 60)
|
||||
return f"{hours}h" if not minutes else f"{hours}h{minutes}m"
|
||||
|
||||
|
||||
def summarize_holders(entries: list[dict[str, Any]]) -> str:
|
||||
"""Compact "who is holding the slots" phrase, e.g. ``desktop x4, cli``."""
|
||||
if not entries:
|
||||
return ""
|
||||
counts: dict[str, int] = {}
|
||||
for entry in entries:
|
||||
surface = str(entry.get("surface") or "unknown")
|
||||
counts[surface] = counts.get(surface, 0) + 1
|
||||
held = ", ".join(
|
||||
f"{surface} x{n}" if n > 1 else surface
|
||||
for surface, n in sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))
|
||||
)
|
||||
started = [t for t in (_optional_float(e.get("started_at")) for e in entries) if t]
|
||||
if started:
|
||||
held += f", oldest {format_age(time.time() - min(started))} ago"
|
||||
return held
|
||||
|
||||
|
||||
def active_session_limit_message(
|
||||
active_count: int,
|
||||
max_sessions: int,
|
||||
entries: Optional[list[dict[str, Any]]] = None,
|
||||
) -> str:
|
||||
# Name the holders: the slots are shared across CLI, desktop/TUI and the
|
||||
# messaging gateway, so the surface that gets rejected is usually NOT the
|
||||
# one squatting on them (idle desktop chats starving a Discord bot, say).
|
||||
# Without this the message is unactionable and the only way to find out is
|
||||
# reading runtime/active_sessions.json by hand.
|
||||
held = summarize_holders(entries or [])
|
||||
detail = f" Held by: {held}." if held else ""
|
||||
return (
|
||||
f"Hermes is at the active session limit ({active_count}/{max_sessions}). "
|
||||
"Try again when another session finishes."
|
||||
f"Hermes is at the active session limit ({active_count}/{max_sessions})."
|
||||
f"{detail} Try again when another session finishes."
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -284,7 +321,9 @@ def try_acquire_active_session(
|
|||
max_sessions,
|
||||
surface,
|
||||
)
|
||||
return None, active_session_limit_message(active_count, max_sessions)
|
||||
return None, active_session_limit_message(
|
||||
active_count, max_sessions, entries
|
||||
)
|
||||
entries.append(entry)
|
||||
_write_entries(state_path, entries)
|
||||
|
||||
|
|
@ -348,6 +387,36 @@ def transfer_active_session(
|
|||
return updated
|
||||
|
||||
|
||||
def release_orphaned_leases(live_lease_ids: set[str]) -> int:
|
||||
"""Drop this process's registry entries that no live session owns.
|
||||
|
||||
``_prune_dead`` only reclaims leases whose owning process died. A server
|
||||
that runs for days (``hermes dashboard`` / ``serve``) never trips that
|
||||
check, so a lease whose session skipped teardown is held until restart.
|
||||
The owning process is the only authority on which of its own leases are
|
||||
real, so it drops the rest itself — exact, with no heartbeat write on the
|
||||
turn path and no staleness threshold to tune.
|
||||
"""
|
||||
pid = os.getpid()
|
||||
state_path = _state_path()
|
||||
# With the cap disabled the registry is never written, so don't take a lock
|
||||
# (or create its file) on the idle-reaper tick for the majority of installs.
|
||||
if not state_path.exists():
|
||||
return 0
|
||||
with _FileLock(_lock_path()):
|
||||
entries = _prune_dead(_read_entries(state_path))
|
||||
kept = [
|
||||
entry
|
||||
for entry in entries
|
||||
if entry.get("pid") != pid
|
||||
or str(entry.get("lease_id") or "") in live_lease_ids
|
||||
]
|
||||
dropped = len(entries) - len(kept)
|
||||
if dropped:
|
||||
_write_entries(state_path, kept)
|
||||
return dropped
|
||||
|
||||
|
||||
def active_session_registry_snapshot() -> list[dict[str, Any]]:
|
||||
"""Return the pruned active-session registry for diagnostics/tests."""
|
||||
state_path = _state_path()
|
||||
|
|
|
|||
|
|
@ -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
|
||||
# =========================================================================
|
||||
|
|
|
|||
|
|
@ -57,10 +57,10 @@ def test_active_session_lease_blocks_until_release(tmp_path, monkeypatch):
|
|||
)
|
||||
|
||||
assert blocked_lease is None
|
||||
assert blocked_message == (
|
||||
"Hermes is at the active session limit (1/1). "
|
||||
"Try again when another session finishes."
|
||||
)
|
||||
assert "active session limit (1/1)" in blocked_message
|
||||
# The rejected surface is rarely the one holding the slots, so the message
|
||||
# must name the holder — here the "cli" lease, not the blocked "tui" one.
|
||||
assert "Held by: cli" in blocked_message
|
||||
|
||||
lease.release()
|
||||
|
||||
|
|
@ -354,3 +354,33 @@ def test_pid_start_time_mismatch_prunes_reused_pid(tmp_path, monkeypatch):
|
|||
"new-session"
|
||||
]
|
||||
lease.release()
|
||||
|
||||
|
||||
def test_release_orphaned_leases_reclaims_only_unowned_own_pid_entries(tmp_path, monkeypatch):
|
||||
"""A long-lived server must reclaim leases whose session skipped teardown.
|
||||
|
||||
``_prune_dead`` only fires when the owning pid dies, so a ``hermes
|
||||
dashboard`` running for days holds a leaked lease until restart. The
|
||||
process reconciles against the leases it still owns instead.
|
||||
"""
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
|
||||
cfg = {"max_concurrent_sessions": 5}
|
||||
kept, orphan = (
|
||||
active_sessions.try_acquire_active_session(
|
||||
session_id=sid, surface="desktop", config=cfg
|
||||
)[0]
|
||||
for sid in ("kept", "orphaned")
|
||||
)
|
||||
# Another live process's lease is not ours to reclaim.
|
||||
active_sessions._write_entries(
|
||||
active_sessions._state_path(),
|
||||
active_sessions._read_entries(active_sessions._state_path())
|
||||
+ [{"lease_id": "elsewhere", "session_id": "other", "surface": "cli", "pid": os.getpid() }],
|
||||
)
|
||||
|
||||
assert active_sessions.release_orphaned_leases({kept.lease_id, "elsewhere"}) == 1
|
||||
assert sorted(
|
||||
entry["session_id"]
|
||||
for entry in active_sessions.active_session_registry_snapshot()
|
||||
) == ["kept", "other"]
|
||||
assert orphan is not None
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
||||
|
|
|
|||
|
|
@ -37,7 +37,7 @@ def _neuter_agent_prewarm_timer(request, monkeypatch):
|
|||
yield
|
||||
|
||||
|
||||
def test_session_create_rejects_at_active_session_limit(monkeypatch, tmp_path):
|
||||
def test_session_slot_is_claimed_on_first_turn_not_on_create(monkeypatch, tmp_path):
|
||||
home = tmp_path / ".hermes"
|
||||
home.mkdir()
|
||||
(home / "config.yaml").write_text("max_concurrent_sessions: 1\n", encoding="utf-8")
|
||||
|
|
@ -56,23 +56,31 @@ def test_session_create_rejects_at_active_session_limit(monkeypatch, tmp_path):
|
|||
monkeypatch.setattr(server, "_start_agent_build", lambda *args, **kwargs: None)
|
||||
monkeypatch.setattr(server, "_completion_cwd", lambda params=None: str(tmp_path))
|
||||
|
||||
# Opening a chat must NOT take a slot. Every tile paint and every
|
||||
# background reconnect-resume calls session.create, and an unprompted
|
||||
# draft has no DB row and is filtered out of the sidebar — so a slot
|
||||
# held here is invisible to the user while still starving the other
|
||||
# surfaces that share this cap.
|
||||
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]
|
||||
assert "result" in first and "result" in second
|
||||
sid = first["result"]["session_id"]
|
||||
other = second["result"]["session_id"]
|
||||
assert active_session_registry_snapshot() == []
|
||||
|
||||
# The first turn is what claims the slot, and is re-entrant.
|
||||
assert server._ensure_active_session_slot(sid, server._sessions[sid]) is None
|
||||
assert server._ensure_active_session_slot(sid, server._sessions[sid]) is None
|
||||
assert len(active_session_registry_snapshot()) == 1
|
||||
|
||||
blocked = server._ensure_active_session_slot(other, server._sessions[other])
|
||||
assert "active session limit (1/1)" in blocked
|
||||
|
||||
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
|
||||
assert server._ensure_active_session_slot(other, server._sessions[other]) is None
|
||||
finally:
|
||||
_clear_server_sessions()
|
||||
server._cfg_cache = None
|
||||
|
|
|
|||
|
|
@ -512,6 +512,33 @@ def _claim_active_session_slot(
|
|||
return None, None
|
||||
|
||||
|
||||
def _ensure_active_session_slot(sid: str, session: dict) -> str | None:
|
||||
"""Claim this session's cap slot on its first real turn; None when ok.
|
||||
|
||||
session.create / session.resume deliberately do NOT claim one. Every
|
||||
desktop tile paint, background reconnect-resume and abandoned draft opens a
|
||||
session just to paint a composer, and a slot held by one of those is
|
||||
invisible everywhere: an unprompted draft has no DB row, and the sidebar
|
||||
filters it out with min_messages=1. Idle desktop tabs therefore silently
|
||||
starved the messaging gateway, which shares this cap — five parked tabs on
|
||||
a websocket-flappy host locked a Discord bot out of a 5-slot cap while
|
||||
running no agents at all. Claiming on the first turn mirrors the lazy
|
||||
contract _ensure_session_db_row already uses for the row itself, and keeps
|
||||
the invariant that anything holding a slot is something the user can see.
|
||||
"""
|
||||
if session.get("active_session_lease") is not None:
|
||||
return None
|
||||
lease, limit_message = _claim_active_session_slot(
|
||||
str(session.get("session_key") or ""),
|
||||
live_session_id=sid,
|
||||
surface=_session_source(session),
|
||||
)
|
||||
if limit_message is not None:
|
||||
return limit_message
|
||||
session["active_session_lease"] = lease
|
||||
return None
|
||||
|
||||
|
||||
def _release_active_session_slot(session: dict | None) -> None:
|
||||
if not session:
|
||||
return
|
||||
|
|
@ -980,6 +1007,24 @@ def _reap_idle_sessions() -> None:
|
|||
for sid in victims:
|
||||
_close_session_by_id(sid, end_reason="idle_timeout")
|
||||
_enforce_session_cap()
|
||||
_reclaim_orphaned_leases()
|
||||
|
||||
|
||||
def _reclaim_orphaned_leases() -> None:
|
||||
"""Hand the registry the lease ids we still own so it can drop the rest."""
|
||||
try:
|
||||
from hermes_cli.active_sessions import release_orphaned_leases
|
||||
|
||||
with _sessions_lock:
|
||||
live = {
|
||||
lease.lease_id
|
||||
for session in _sessions.values()
|
||||
if (lease := session.get("active_session_lease")) is not None
|
||||
}
|
||||
if dropped := release_orphaned_leases(live):
|
||||
logger.info("Reclaimed %d orphaned active-session lease(s)", dropped)
|
||||
except Exception:
|
||||
logger.debug("orphaned lease reclaim failed", exc_info=True)
|
||||
|
||||
|
||||
# Soft LRU cap on in-memory sessions. The 6h TTL reaper above only frees
|
||||
|
|
@ -6997,11 +7042,7 @@ def _(rid, params: dict) -> dict:
|
|||
|
||||
ready = threading.Event()
|
||||
now = time.time()
|
||||
lease, limit_message = _claim_active_session_slot(
|
||||
key, live_session_id=sid, surface=source
|
||||
)
|
||||
if limit_message is not None:
|
||||
return _err(rid, 4090, limit_message)
|
||||
lease = None # claimed lazily on the first turn (_ensure_active_session_slot)
|
||||
|
||||
with _sessions_lock:
|
||||
_sessions[sid] = {
|
||||
|
|
@ -7445,11 +7486,7 @@ def _(rid, params: dict) -> dict:
|
|||
if is_truthy_value(params.get("lazy", False)):
|
||||
sid = uuid.uuid4().hex[:8]
|
||||
source = _resolve_session_source(str(params.get("source") or "").strip() or None)
|
||||
lease, limit_message = _claim_active_session_slot(
|
||||
target, live_session_id=sid, surface=source
|
||||
)
|
||||
if limit_message is not None:
|
||||
return _err(rid, 4090, limit_message)
|
||||
lease = None # claimed lazily on the first turn (_ensure_active_session_slot)
|
||||
try:
|
||||
db.reopen_session(target)
|
||||
# The child's OWN conversation only — include_ancestors would prepend
|
||||
|
|
@ -7526,11 +7563,7 @@ def _(rid, params: dict) -> dict:
|
|||
if not is_truthy_value(params.get("eager_build", False)):
|
||||
sid = uuid.uuid4().hex[:8]
|
||||
source = _resolve_session_source(str(params.get("source") or "").strip() or None)
|
||||
lease, limit_message = _claim_active_session_slot(
|
||||
target, live_session_id=sid, surface=source
|
||||
)
|
||||
if limit_message is not None:
|
||||
return _err(rid, 4090, limit_message)
|
||||
lease = None # claimed lazily on the first turn (_ensure_active_session_slot)
|
||||
# Interactive resume routes approvals/clarify through gateway prompts;
|
||||
# the deferred build wires the remaining per-session callbacks.
|
||||
_enable_gateway_prompts()
|
||||
|
|
@ -7605,11 +7638,7 @@ def _(rid, params: dict) -> dict:
|
|||
# dispatch thread (it's not a _LONG_HANDLER), blocking fast-path RPCs.
|
||||
sid = uuid.uuid4().hex[:8]
|
||||
source = _resolve_session_source(str(params.get("source") or "").strip() or None)
|
||||
lease, limit_message = _claim_active_session_slot(
|
||||
target, live_session_id=sid, surface=source
|
||||
)
|
||||
if limit_message is not None:
|
||||
return _err(rid, 4090, limit_message)
|
||||
lease = None # claimed lazily on the first turn (_ensure_active_session_slot)
|
||||
_enable_gateway_prompts()
|
||||
home_token = (
|
||||
set_hermes_home_override(str(profile_home)) if profile_home is not None else None
|
||||
|
|
@ -10423,11 +10452,7 @@ def _(rid, params: dict) -> dict:
|
|||
new_key = _new_session_key()
|
||||
new_sid = uuid.uuid4().hex[:8]
|
||||
source = _session_source(session)
|
||||
lease, limit_message = _claim_active_session_slot(
|
||||
new_key, live_session_id=new_sid, surface=source
|
||||
)
|
||||
if limit_message is not None:
|
||||
return _err(rid, 4090, limit_message)
|
||||
lease = None # claimed lazily on the first turn (_ensure_active_session_slot)
|
||||
branch_name = params.get("name", "")
|
||||
try:
|
||||
if branch_name:
|
||||
|
|
@ -10932,6 +10957,8 @@ def _(rid, params: dict) -> dict:
|
|||
session, err = _sess_nowait(params, rid)
|
||||
if err:
|
||||
return err
|
||||
if (limit_message := _ensure_active_session_slot(sid, session)) is not None:
|
||||
return _err(rid, 4090, limit_message)
|
||||
if truncate_user_ordinal is not None and isinstance(text, str):
|
||||
# A rewind/regenerate replays a turn from what the transcript shows. A
|
||||
# skill turn shows its invocation, so re-expand it here — otherwise
|
||||
|
|
|
|||
|
|
@ -1808,8 +1808,14 @@ and messaging gateway:
|
|||
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.
|
||||
A slot is taken when a session runs its **first turn**, not when a chat window
|
||||
is opened. Opening, resuming or reconnecting to a chat costs nothing until you
|
||||
send a message, so idle desktop tabs (and the background resumes a flaky
|
||||
websocket triggers) cannot starve the messaging gateway that shares this cap.
|
||||
|
||||
When the cap is reached, Hermes returns a direct limit message naming which
|
||||
surfaces hold the slots. Existing active sessions keep their normal behavior.
|
||||
Run `hermes status` to see the current slot usage and every holder.
|
||||
|
||||
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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue