fix(gateway): a queued turn-end lease release is in effect for the next claim at once

The turn-end active-session release moved to a background worker, while the next inbound
claim (_claim_active_session_slot) runs on the loop straight after. try_acquire counted the
finished turn's still-written registry entry, so with max_concurrent_sessions set the next
session was refused "at the active session limit (1/1)" with nothing running (6/100 and 2/100
trials vs 0/100 on main), and a Desktop/TUI writer taking over the same session right after a
gateway turn would see it "already owned".

Only the durable file write stays deferred now: release_soon marks the lease queued before it
returns, and every in-process registry read (_read_entries: claims, the Desktop snapshot, the
liveness guards, the orphan sweep) hides queued lease ids, so the next locked write persists
the drop. The release worker unmarks the lease under the registry lock before dropping it.
Review A's 100-trial repro: REFUSED 0/100.

gateway/active_lease_io.py

@@ -2,10 +2,12 @@
 
 ``ActiveSessionLease.release()`` takes the registry's file lock, re-reads and prunes the JSON (a
 liveness probe per entry) and atomically rewrites it. Every turn end ran it on the loop, a measured
-recurring stall with many turns finishing. Releases go to ONE worker so they land in the order the
-turns ended. A release that lands after the same session's next acquire is harmless: the registry
-drops entries by lease id and that acquire re-keyed the entry to its own id. The lease pinned its
-registry path at acquisition, so the worker needs no profile scope.
+recurring stall with many turns finishing. Only that file write is deferred: the lease leaves this
+process's view of the registry before ``release_soon`` returns (``mark_release_queued``), so the
+next claim on the loop -- another session at ``max_concurrent_sessions``, or a Desktop/TUI writer
+taking over this session -- sees the slot free, exactly as when the release ran inline. Releases
+go to ONE worker so the durable drops land in turn-end order. The lease pinned its registry path
+at acquisition, so the worker needs no profile scope.
 """
 
 from __future__ import annotations
@@ -21,8 +23,11 @@ _pool_lock = threading.Lock()
 
 
 def release_soon(lease) -> None:
-    """Queue *lease*'s release; returns at once."""
+    """Retire *lease* from this process's registry view now and queue its durable release."""
     global _pool
+    from hermes_cli.active_sessions import mark_release_queued
+
+    mark_release_queued(lease)
     with _pool_lock:
         if _pool is None:
             _pool = concurrent.futures.ThreadPoolExecutor(max_workers=1, thread_name_prefix="active-lease-io")

hermes_cli/active_sessions.py

@@ -12,6 +12,7 @@ import logging
 import collections
 import math
 import os
+import threading
 import time
 import uuid
 from contextlib import contextmanager, suppress
@@ -250,7 +251,7 @@ def _read_entries(path: Path, *, strict: bool = False) -> list[dict[str, Any]]:
         return []
     valid = [entry for entry in entries if isinstance(entry, dict)]
     if not strict:
-        return valid
+        return _without_queued_releases(path, valid)
     if len(valid) != len(entries):
         raise invalid("contains invalid entries")
     seen_leases: set[str] = set()
@@ -271,7 +272,50 @@ def _read_entries(path: Path, *, strict: bool = False) -> list[dict[str, Any]]:
             if bad():
                 raise invalid(f"contains {what}")
         seen_leases.add(lease_id)
-    return valid
+    return _without_queued_releases(path, valid)
+
+
+# Leases this process has queued for release (gateway/active_lease_io.py) whose durable drop has not
+# landed yet, keyed by registry file. Every in-process read hides them, so a claim, count or
+# ownership check made right after a turn end sees the registry as if the release had already run
+# (a slot freed, the session unowned); the next locked write persists the drop. Other processes
+# still see the entry until the release worker rewrites the file.
+_queued_releases: dict[str, set[str]] = {}
+_queued_releases_lock = threading.Lock()
+
+
+def _registry_key(path: Path) -> str:
+    return os.path.normcase(os.path.abspath(path))
+
+
+def mark_release_queued(lease: ActiveSessionLease) -> None:
+    """Retire ``lease`` from this process's view of the registry now; its file write comes later."""
+    if not lease.enabled or lease.released:
+        return
+    state_path, _ = _lease_paths(lease)
+    with _queued_releases_lock:
+        _queued_releases.setdefault(_registry_key(state_path), set()).add(lease.lease_id)
+
+
+def _unmark_release_queued(state_path: Path, lease_id: str) -> None:
+    """Make ``lease_id`` visible again to the caller holding the registry lock, which drops it."""
+    key = _registry_key(state_path)
+    with _queued_releases_lock:
+        queued = _queued_releases.get(key)
+        if queued is not None:
+            queued.discard(lease_id)
+            if not queued:
+                del _queued_releases[key]
+
+
+def _without_queued_releases(path: Path, entries: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    if not _queued_releases:
+        return entries
+    with _queued_releases_lock:
+        queued = set(_queued_releases.get(_registry_key(path), ()))
+    if not queued:
+        return entries
+    return [e for e in entries if str(e.get("lease_id") or "") not in queued]
 
 
 def _nonblank_str(v: Any) -> bool:
@@ -585,6 +629,7 @@ def release_active_session(lease: ActiveSessionLease) -> None:
     # See #85431.
     state_path, lock_path = _lease_paths(lease)
     with _FileLock(lock_path):
+        _unmark_release_queued(state_path, lease.lease_id)
         if lease.released:
             return
         loaded = _read_live_entries(
@@ -780,6 +825,7 @@ def release_active_session_liveness_guard(
 
     state_path, lock_path = _lease_paths(lease)
     with _FileLock(lock_path):
+        _unmark_release_queued(state_path, lease.lease_id)
         entries = _prune_dead(
             _read_entries(state_path, strict=True), strict=True, target_session_id=session_id,
         )

tests/gateway/test_active_lease_release_order.py

@@ -0,0 +1,98 @@
+"""A turn end's queued lease release is already in effect for the next claim in this process.
+
+The durable registry drop runs on a worker (gateway/active_lease_io.py), but the finished turn's
+slot must be free the moment the turn ends: the next session's claim at max_concurrent_sessions,
+and a Desktop writer taking over the same session, run on the loop straight after. The worker is
+held busy with an earlier release (registry contention) so its drop provably has not landed.
+"""
+
+import threading
+
+import gateway.run as run_mod
+from gateway.active_lease_io import release_soon
+from gateway.config import Platform
+from gateway.session import SessionSource
+from hermes_cli import active_sessions
+
+
+class _ReleaseStuckOnTheRegistry:
+    """An earlier queued release that holds the single release worker until told to finish."""
+
+    enabled = False
+    released = False
+
+    def __init__(self):
+        self.started = threading.Event()
+        self.finish = threading.Event()
+
+    def release(self):
+        self.started.set()
+        self.finish.wait(10)
+
+
+def _runner(cfg):
+    runner = run_mod.GatewayRunner.__new__(run_mod.GatewayRunner)
+    runner._running_agents = {}
+    runner._running_agents_ts = {}
+    runner._busy_ack_ts = {}
+    runner.config = cfg
+    return runner
+
+
+def _end_gateway_turn(runner, session_key, source):
+    lease, refusal = runner._claim_active_session_slot(session_key, source)
+    assert refusal is None and lease is not None, refusal
+    runner._session_state(session_key).turn.lease = lease
+    runner._release_running_agent_state(session_key)
+    return lease
+
+
+def _registry_session_ids(home):
+    return sorted(e["session_id"] for e in active_sessions._read_entries(
+        home / "runtime" / "active_sessions.json", strict=True))
+
+
+def _drain(lease, stuck):
+    stuck.finish.set()
+    done = threading.Event()
+    release_soon(type("Marker", (), {"enabled": False, "released": False, "release": lambda self: done.set()})())
+    assert done.wait(10), "the release worker never drained"
+    assert lease.released
+
+
+def test_the_next_sessions_claim_at_the_cap_sees_the_ended_turns_slot_free(tmp_path, monkeypatch):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    runner = _runner({"max_concurrent_sessions": 1})
+    src = SessionSource(platform=Platform.TELEGRAM, chat_id="c1", user_id="u1")
+    stuck = _ReleaseStuckOnTheRegistry()
+    release_soon(stuck)
+    assert stuck.started.wait(10)
+
+    lease_a = _end_gateway_turn(runner, "agent:main:telegram:dm:a", src)
+    lease_b, refusal = runner._claim_active_session_slot("agent:main:telegram:dm:b", src)
+    assert refusal is None, f"the next session was refused with nothing running: {refusal}"
+
+    _drain(lease_a, stuck)
+    assert _registry_session_ids(tmp_path) == ["agent:main:telegram:dm:b"]
+    lease_b.release()
+
+
+def test_a_desktop_claim_of_the_session_a_gateway_turn_just_ended_is_not_already_owned(tmp_path, monkeypatch):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    from tui_gateway import server
+
+    runner = _runner({})
+    src = SessionSource(platform=Platform.TELEGRAM, chat_id="c1", user_id="u1")
+    key = "agent:main:telegram:dm:shared"
+    stuck = _ReleaseStuckOnTheRegistry()
+    release_soon(stuck)
+    assert stuck.started.wait(10)
+
+    lease_gw = _end_gateway_turn(runner, key, src)
+    lease_desk, refusal = server._claim_active_session_slot(key, live_session_id="desktop-view", surface="desktop", profile_home=tmp_path)
+    assert refusal is None, f"Desktop saw the ended gateway turn as the session's owner: {refusal}"
+
+    _drain(lease_gw, stuck)
+    entries = active_sessions._read_entries(tmp_path / "runtime" / "active_sessions.json", strict=True)
+    assert [(e["session_id"], e["surface"]) for e in entries] == [(key, "desktop")]
+    lease_desk.release()