From f4261c229b4a1572bb1fe0f0cfdb5dbc9362edca Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 20:18:31 -0700 Subject: [PATCH] refactor(hermes_cli): compact active_sessions lease registry (775->~695 LOC), unify lease-drop/session-held helpers --- hermes_cli/active_sessions.py | 236 ++++++++++++---------------------- 1 file changed, 81 insertions(+), 155 deletions(-) diff --git a/hermes_cli/active_sessions.py b/hermes_cli/active_sessions.py index 015b8a636f..a375f8567a 100644 --- a/hermes_cli/active_sessions.py +++ b/hermes_cli/active_sessions.py @@ -1,6 +1,6 @@ """Cross-process active chat session leases. -The session database records persisted conversations. This module records +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. """ @@ -31,14 +31,9 @@ def coerce_max_concurrent_sessions(value: Any, key: str = "max_concurrent_sessio """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, bool): + raise ValueError(value) if isinstance(value, float): if not value.is_integer(): raise ValueError(value) @@ -49,14 +44,10 @@ def coerce_max_concurrent_sessions(value: Any, key: str = "max_concurrent_sessio parsed = int(value) except (TypeError, ValueError): logger.warning( - "Ignoring invalid %s=%r (expected a positive integer; 0/null disables)", - key, - value, + "Ignoring invalid %s=%r (expected a positive integer; 0/null disables)", key, value ) return None - if parsed <= 0: - return None - return parsed + return parsed if parsed > 0 else None def resolve_max_concurrent_sessions(config: Any) -> Optional[int]: @@ -103,9 +94,7 @@ def summarize_holders(entries: list[dict[str, Any]]) -> str: def active_session_limit_message( - active_count: int, - max_sessions: int, - entries: Optional[list[dict[str, Any]]] = None, + active_count: int, max_sessions: int, entries: Optional[list[dict[str, Any]]] = None ) -> str: # Name the holders: slots are shared across CLI, desktop/TUI and gateway, # so the rejected surface is usually NOT the one squatting on them. @@ -117,10 +106,6 @@ def active_session_limit_message( ) -def _registry_home(registry_home: str | Path | None = None) -> Path: - return Path(registry_home) if registry_home is not None else Path(get_hermes_home()) - - # Machine-readable refusal reasons: the reason is the contract, the message is # for people. Capacity = "busy, come back later"; ownership = "this session # has a live owner and writing would interleave with theirs". @@ -161,9 +146,7 @@ def _is_same_writer(entry: dict[str, Any], metadata: Optional[dict[str, Any]]) - return False existing_live = str((entry.get("metadata") or {}).get("live_session_id") or "") incoming_live = str((metadata or {}).get("live_session_id") or "") - if not existing_live or not incoming_live: - return False - return existing_live == incoming_live + return bool(existing_live and incoming_live) and existing_live == incoming_live def session_already_owned_message(session_id: str, entry: dict[str, Any]) -> str: @@ -178,21 +161,20 @@ def session_already_owned_message(session_id: str, entry: dict[str, Any]) -> str ) -def _state_dir(registry_home: str | Path | None = None) -> Path: - return _registry_home(registry_home) / "runtime" +def _registry_home(registry_home: str | Path | None = None) -> Path: + return Path(registry_home) if registry_home is not None else Path(get_hermes_home()) def _state_path(registry_home: str | Path | None = None) -> Path: - return _state_dir(registry_home) / "active_sessions.json" + return _registry_home(registry_home) / "runtime" / "active_sessions.json" def _lock_path(registry_home: str | Path | None = None) -> Path: - return _state_dir(registry_home) / "active_sessions.lock" + return _registry_home(registry_home) / "runtime" / "active_sessions.lock" def _lease_paths( - lease: Optional["ActiveSessionLease"] = None, - registry_home: str | Path | None = None, + lease: Optional["ActiveSessionLease"] = None, registry_home: str | Path | None = None ) -> tuple[Path, Path]: if lease is not None and lease.state_path is not None and lease.lock_path is not None: return lease.state_path, lease.lock_path @@ -243,6 +225,9 @@ class _FileLock: def _read_entries(path: Path, *, strict: bool = False) -> list[dict[str, Any]]: + def invalid(what: str) -> ActiveSessionRegistryError: + return ActiveSessionRegistryError(f"active session registry {what}: {path}") + try: with open(path, "r", encoding="utf-8") as fh: data = json.load(fh) @@ -250,25 +235,19 @@ def _read_entries(path: Path, *, strict: bool = False) -> list[dict[str, Any]]: return [] except Exception as exc: if strict: - raise ActiveSessionRegistryError( - f"active session registry unreadable: {path}" - ) from exc + raise invalid("unreadable") from exc 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): if strict: - raise ActiveSessionRegistryError( - f"active session registry has invalid shape: {path}" - ) + raise invalid("has invalid shape") return [] valid = [entry for entry in entries if isinstance(entry, dict)] if not strict: return valid if len(valid) != len(entries): - raise ActiveSessionRegistryError( - f"active session registry contains invalid entries: {path}" - ) + raise invalid("contains invalid entries") seen_leases: set[str] = set() for entry in valid: lease_id = entry.get("lease_id") @@ -285,9 +264,7 @@ def _read_entries(path: Path, *, strict: bool = False) -> list[dict[str, Any]]: (lambda: not _valid_process_start(entry.get("process_start_time")), "an invalid process start time"), ): if bad(): - raise ActiveSessionRegistryError( - f"active session registry contains {what}: {path}" - ) + raise invalid(f"contains {what}") seen_leases.add(lease_id) return valid @@ -357,18 +334,19 @@ def _pid_liveness(pid: Any, process_start_time: Any = None, *, lenient: bool = F ``lenient`` never returns None: an unparseable pid or a failed existence probe counts as dead, an unreadable current start time as alive. """ + unknown_dead = False if lenient else None try: pid_int = int(pid) except (TypeError, ValueError): - return False if lenient else None + return unknown_dead if pid_int <= 0: - return False if lenient else None + return unknown_dead try: from gateway.status import _pid_exists exists = bool(_pid_exists(pid_int)) except Exception: - return False if lenient else None + return unknown_dead if not exists: return False expected_start = _optional_float(process_start_time) @@ -380,26 +358,17 @@ def _pid_liveness(pid: Any, process_start_time: Any = None, *, lenient: bool = F return abs(current_start - expected_start) < 0.001 -def _pid_alive(pid: Any, process_start_time: Any = None) -> bool: - return bool(_pid_liveness(pid, process_start_time, lenient=True)) - - -def _prune_dead( - entries: list[dict[str, Any]], *, strict: bool = False -) -> list[dict[str, Any]]: +def _prune_dead(entries: list[dict[str, Any]], *, strict: bool = False) -> list[dict[str, Any]]: + """Keep entries whose owner is alive; tracked/strict entries must be provably so.""" live: list[dict[str, Any]] = [] for entry in entries: - tracked = bool(entry.get("track_liveness")) - if strict or tracked: - state = _pid_liveness(entry.get("pid"), entry.get("process_start_time")) - if state is None: - raise ActiveSessionRegistryError( - "active session owner liveness is unknown" - ) - if state: - live.append(entry) - continue - if _pid_alive(entry.get("pid"), entry.get("process_start_time")): + tracked = strict or bool(entry.get("track_liveness")) + state = _pid_liveness( + entry.get("pid"), entry.get("process_start_time"), lenient=not tracked + ) + if state is None: + raise ActiveSessionRegistryError("active session owner liveness is unknown") + if state: live.append(entry) return live @@ -428,8 +397,19 @@ def _clean_metadata(metadata: dict[str, Any]) -> dict[str, Any]: return {str(k): v for k, v in metadata.items() if isinstance(k, str)} -def _without_lease(entries: list[dict[str, Any]], lease_id: str) -> list[dict[str, Any]]: - return [e for e in entries if str(e.get("lease_id") or "") != lease_id] +def _drop_lease( + state_path: Path, entries: list[dict[str, Any]], lease_id: str +) -> list[dict[str, Any]]: + """Remove ``lease_id`` from ``entries``, writing the registry only if it was present.""" + kept = [e for e in entries if str(e.get("lease_id") or "") != lease_id] + if len(kept) != len(entries): + _write_entries(state_path, kept) + return kept + + +def _holds_session(entries: list[dict[str, Any]], session_id: str) -> bool: + target = str(session_id or "") + return bool(target) and any(str(e.get("session_id") or "") == target for e in entries) def _read_live_entries( @@ -503,28 +483,17 @@ def try_acquire_active_session( # refuses empty session ids): hand back a no-op lease. if not key and not track_liveness: return ActiveSessionLease( - lease_id=lease_id, - session_id=key, - surface=str(surface), - enabled=False, + lease_id=lease_id, session_id=key, surface=str(surface), enabled=False ), None entry = _lease_entry( - lease_id=lease_id, - session_id=key, - surface=str(surface), - metadata=metadata, + lease_id=lease_id, session_id=key, surface=str(surface), metadata=metadata, track_liveness=track_liveness, ) - state_path, lock_path = _lease_paths(registry_home=registry_home) lease = ActiveSessionLease( - lease_id=lease_id, - session_id=key, - surface=str(surface), - state_path=state_path, - lock_path=lock_path, - track_liveness=track_liveness, + lease_id=lease_id, session_id=key, surface=str(surface), state_path=state_path, + lock_path=lock_path, track_liveness=track_liveness, ) with _FileLock(lock_path): # A capacity cap could degrade open; exclusivity cannot: "could not @@ -536,11 +505,9 @@ def try_acquire_active_session( ) if loaded is None: return None, ActiveSessionRefusal( - ( - "Hermes could not read the active-session registry at " - f"{state_path}, so it cannot prove this session has no other " - "live owner. Fix or remove that file and try again." - ), + "Hermes could not read the active-session registry at " + f"{state_path}, so it cannot prove this session has no other " + "live owner. Fix or remove that file and try again.", SESSION_COORDINATION_UNAVAILABLE, ) raw_entries, entries = loaded @@ -555,7 +522,6 @@ def try_acquire_active_session( for index, existing in enumerate(entries): if str(existing.get("session_id") or "") != key: continue - # The same writer is not a second writer: a live session that # leaked its lease reference would otherwise be fenced out of # its own session permanently (pruning only removes entries @@ -564,34 +530,27 @@ def try_acquire_active_session( entries[index] = entry _write_entries(state_path, entries) return lease, None - _write_entries(state_path, entries) logger.info( "Refused active session %s: already held by pid=%s surface=%s", - key, - existing.get("pid"), - existing.get("surface"), + key, existing.get("pid"), existing.get("surface"), ) return None, ActiveSessionRefusal( - session_already_owned_message(key, existing), - SESSION_NOT_OWNED, + session_already_owned_message(key, existing), SESSION_NOT_OWNED ) # Capacity second, and only when an operator asked for one. - if max_sessions is not None: + if max_sessions is not None and len(entries) >= max_sessions: 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, ActiveSessionRefusal( - active_session_limit_message(active_count, max_sessions, entries), - MAX_CONCURRENT_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, ActiveSessionRefusal( + active_session_limit_message(active_count, max_sessions, entries), + MAX_CONCURRENT_SESSIONS, + ) entries.append(entry) _write_entries(state_path, entries) @@ -600,7 +559,7 @@ def try_acquire_active_session( def release_active_session(lease: ActiveSessionLease) -> None: # Prefer the registry the lease was acquired against: the caller may be - # running under a profile HERMES_HOME override (#85431). + # running under a profile HERMES_HOME override. state_path, lock_path = _lease_paths(lease) with _FileLock(lock_path): if lease.released: @@ -611,24 +570,16 @@ def release_active_session(lease: ActiveSessionLease) -> None: "releasing an untracked lease", ) if loaded is not None: - entries = loaded[1] - kept = _without_lease(entries, lease.lease_id) - if len(kept) != len(entries): - _write_entries(state_path, kept) + _drop_lease(state_path, loaded[1], lease.lease_id) lease.released = True def transfer_active_session( - lease: ActiveSessionLease, - *, - session_id: str, - metadata: Optional[dict[str, Any]] = None, + lease: ActiveSessionLease, *, session_id: str, metadata: Optional[dict[str, Any]] = None ) -> bool: """Move an existing lease to a new session id without dropping the slot.""" new_session_id = str(session_id or "") - if not new_session_id: - return False - if lease.released: + if not new_session_id or lease.released: return False if not lease.enabled: lease.session_id = new_session_id @@ -659,15 +610,10 @@ def transfer_active_session( updated = True break if not updated and lease.track_liveness: - entries.append( - _lease_entry( - lease_id=lease.lease_id, - session_id=new_session_id, - surface=lease.surface, - metadata=metadata, - track_liveness=True, - ) - ) + entries.append(_lease_entry( + lease_id=lease.lease_id, session_id=new_session_id, surface=lease.surface, + metadata=metadata, track_liveness=True, + )) updated = True if updated: _write_entries(state_path, entries) @@ -698,10 +644,8 @@ def release_orphaned_leases(live_lease_ids: set[str]) -> int: return 0 entries = loaded[1] kept = [ - entry - for entry in entries - if entry.get("pid") != pid - or str(entry.get("lease_id") or "") in live_lease_ids + 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: @@ -724,52 +668,34 @@ def active_session_registry_snapshot( @contextmanager def active_session_liveness_guard( - session_id: str, - *, - registry_home: str | Path | None = None, + session_id: str, *, registry_home: str | Path | None = None ) -> Iterator[bool]: """Hold the registry lock while reporting whether ``session_id`` is leased, so no new backend can acquire a lease between the check and the caller's ``end_session`` write.""" - target = str(session_id or "") state_path, lock_path = _lease_paths(registry_home=registry_home) with _FileLock(lock_path): entries = _prune_dead(_read_entries(state_path, strict=True), strict=True) _write_entries(state_path, entries) - yield bool(target) and any( - str(entry.get("session_id") or "") == target for entry in entries - ) + yield _holds_session(entries, session_id) @contextmanager def release_active_session_liveness_guard( - lease: ActiveSessionLease, - session_id: str, + lease: ActiveSessionLease, session_id: str ) -> Iterator[bool]: """Remove ``lease`` and hold its registry lock through a lifecycle write, making cleanup one atomic ownership decision (release, check siblings, end the durable row) before any new backend can reopen it.""" if not lease.enabled or lease.released: - with active_session_liveness_guard( - session_id, registry_home=_registry_home_for_lease(lease) - ) as active: + home = lease.state_path.parent.parent if lease.state_path is not None else None + with active_session_liveness_guard(session_id, registry_home=home) as active: yield active return - target = str(session_id or "") state_path, lock_path = _lease_paths(lease) with _FileLock(lock_path): entries = _prune_dead(_read_entries(state_path, strict=True), strict=True) - kept = _without_lease(entries, lease.lease_id) - if len(kept) != len(entries): - _write_entries(state_path, kept) + kept = _drop_lease(state_path, entries, lease.lease_id) lease.released = True - yield bool(target) and any( - str(entry.get("session_id") or "") == target for entry in kept - ) - - -def _registry_home_for_lease(lease: ActiveSessionLease) -> Path | None: - if lease.state_path is None: - return None - return lease.state_path.parent.parent + yield _holds_session(kept, session_id)