fix(sessions): claim the cap slot on first turn, not on open

An open chat window took a session-cap slot at session.create/resume time.
Every desktop tile paint and every background reconnect-resume opens one, so
on a websocket-flappy host they accumulated: five parked desktop tabs filled a
5-slot cap and locked the messaging gateway (which shares the cap) out for
fourteen minutes while running no agents at all.

A slot held that way is invisible everywhere. An unprompted draft has no DB row
and the sidebar filters it out with min_messages=1, so the only way to diagnose
it was reading runtime/active_sessions.json by hand.

Claim on the first turn instead, mirroring the lazy contract
_ensure_session_db_row already uses for the row itself. Capacity now means an
agent can run rather than that a window exists, and anything holding a slot is
something the user can see.

Also reclaim leases whose session skipped teardown. _prune_dead only fires when
the owning pid dies, and a dashboard/serve backend runs for days, so a leaked
lease was held until restart. The owning process reconciles against the leases
it still holds, which is exact and needs no heartbeat write on the turn path.

hermes_cli/active_sessions.py

@@ -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()

tests/hermes_cli/test_active_sessions.py

@@ -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

tests/test_tui_gateway_server.py

@@ -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

tui_gateway/server.py

@@ -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

website/docs/user-guide/configuration.md

@@ -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