"""SessionStore storage plumbing: per-profile SessionDB handle resolution and the routing-index load/save paths (state.db gateway_routing primary, sessions.json legacy mirror). Mixin split out of ``gateway/session.py``; bound onto ``SessionStore`` via the MRO.""" from __future__ import annotations import contextlib import logging import json import threading from pathlib import Path from typing import TYPE_CHECKING, Any, Dict, Optional from utils import atomic_json_write if TYPE_CHECKING: from gateway.session import SessionEntry # Log-record parity with the origin module. logger = logging.getLogger("gateway.session") # "No SessionDB pinned" sentinel: lets ``_db`` distinguish "resolve from the active scope" from a # deliberate ``store._db = None`` (JSONL fallback). _DB_UNPINNED = object() # Self-documenting sentinel written first into sessions.json; "_" keys are skipped on load. _SESSIONS_JSON_README = ( "LEGACY MIRROR of the gateway routing index (the primary copy lives in the gateway_routing " "table in ~/.hermes/state.db). Maps messaging session keys (agent:main::...) to " "active session IDs. This is NOT the session list. ALL sessions (CLI, TUI, and gateway) live " "in ~/.hermes/state.db and are shown by `hermes sessions list` and `/sessions`. Disable this " "file with `gateway.write_sessions_json: false` in config.yaml." ) def _is_live_system_guard(exc: BaseException) -> bool: """Test-isolation guard: must stay a loud failure and is never cached.""" return isinstance(exc, RuntimeError) and "live-system guard" in str(exc) class SessionPersistenceMixin: """SessionStore storage plumbing: SessionDB handle resolution and routing-index load/save.""" def _open_session_db_for_active_scope(self, db_path: Optional[Path] = None): """SessionDB for the active profile scope. ``db_path`` pins the store; otherwise ``_default_db_path()`` follows the context-local HERMES_HOME (resolved per call so multiplexed profiles reach their own store). Handles are cached per path; failed opens enter a bounded backoff during which callers keep using the JSONL fallback. Resolving here rather than once in ``__init__`` is the whole fix for #88532: it lets the scoping that the multiplexed inbound path already performs actually reach session storage. """ from hermes_state import _default_db_path from hermes_state_registry import acquire path = Path(db_path) if db_path is not None else Path(_default_db_path()) def _open(): try: return acquire(path) # process-wide registry: one writer per path except Exception as e: if not _is_live_system_guard(e): print(f"[gateway] Warning: SQLite session store unavailable, falling back to JSONL: {e}") raise return self._db_handle_cache.get(path, _open, non_cacheable=_is_live_system_guard) def _pinned_db(self): """Return the explicitly pinned DB (``store._db = x``), else ``_DB_UNPINNED``.""" return getattr(self, "_db_pinned", _DB_UNPINNED) @property def _db(self): """The SessionDB for the active profile scope, or a pinned override. Assigning ``store._db`` pins that value for every subsequent read (tests install a fake or disable the DB with ``store._db = None``); unpinned, each read resolves the scope so a multiplexed profile's writes reach its own store.""" pinned = self._pinned_db() return self._open_session_db_for_active_scope() if pinned is _DB_UNPINNED else pinned @_db.setter def _db(self, value) -> None: self._db_pinned = value @property def _routing_db(self): """The one store that owns the routing index, whatever scope is active. ``_entries`` is one flat dict holding every profile's keys, so it must persist to ONE file (``_routing_home``), not whichever profile is scoped — otherwise a mid-turn rewrite and the unscoped startup load see different copies and crash markers under a secondary profile go unrecovered. A pinned handle still wins; bare test instances lacking the handle cache report no DB. Reading it through ``_db`` made that file whichever profile happened to be scoped at the time: a whole-index rewrite during one profile's turn copied every other profile's routing rows into that profile's store, and startup — which runs unscoped — then loaded a different copy than the one the last writer produced. See #66887. """ pinned = self._pinned_db() if pinned is not _DB_UNPINNED: return pinned home = getattr(self, "_routing_home", None) try: if home is None: return self._db return self._open_session_db_for_active_scope(db_path=home / "state.db") except Exception: return None def _named_profile_for_key(self, session_key: Optional[str]) -> Optional[str]: """The non-default profile that owns *session_key*, or None (ambient store is authoritative: multiplexing off, or legacy ``agent:main``). Deliberately does NOT cover "that profile has no directory" — ownership and resolvability are separate questions for ``_db_for_key``.""" if not getattr(self.config, "multiplex_profiles", False): return None profile = self._profile_from_session_key(session_key) return None if not profile or profile == "default" else profile def _profile_home_for_key(self, session_key: Optional[str]) -> Optional[Path]: """HERMES_HOME of the profile owning *session_key*, or None (no named owner or unresolvable).""" profile = self._named_profile_for_key(session_key) if profile is None: return None cache = self._profile_home_cache if profile in cache: return cache[profile] home: Optional[Path] = None try: from hermes_cli.profiles import get_profile_dir, profile_exists if profile_exists(profile): home = Path(get_profile_dir(profile)) except Exception as exc: logger.debug("Could not resolve profile home for %r: %s", session_key, exc) home = None # Only hits are memoized: a profile directory can be provisioned *after* startup (enrollment # bridge), and a cached miss would pin that profile's rows to the ambient store for life. if home is not None: cache[profile] = home return home def _db_for_key(self, session_key: Optional[str]): """The SessionDB holding *session_key*'s rows, whatever scope is active (the owning profile is encoded in the key). ``_db`` follows the ambient HERMES_HOME that only the inbound message path installs; unscoped background work (expiry watcher) would otherwise write profile rows into the ROOT store until the stale-route self-heal drops a live conversation. Background work runs unscoped while operating on every profile's keys out of the single process-wide ``_entries`` dict — ``_session_housekeeping_watcher`` is the clearest case — so it reads and writes the ROOT store for rows that actually live under ``profiles//state.db``. The two writers then drift apart on the same logical session until the routing index disagrees with the row and the #54878 self-heal drops a live conversation (#66887). """ pinned = self._pinned_db() if pinned is not _DB_UNPINNED: return pinned profile = self._named_profile_for_key(session_key) if profile is None: # Default-profile (``agent:main``) rows belong to the launch home, not to whichever # profile's scope happens to be active: a scoped drain tick or cron mirror touching a # default chat used to write its rows into the secondary's store (#102157's picture). routing_home = getattr(self, "_routing_home", None) if routing_home is None or not getattr(self.config, "multiplex_profiles", False): return self._db try: return self._open_session_db_for_active_scope(db_path=routing_home / "state.db") except Exception: return None home = self._profile_home_for_key(session_key) if home is None: # Falling back to the ambient store would split ONE session identity across two # physical stores — fail closed; callers already handle a missing DB. logger.warning( "gateway.session: profile %r has no resolvable home (key %r); refusing to fall " "back to the ambient store", profile, session_key) return None try: return self._open_session_db_for_active_scope(db_path=home / "state.db") except Exception: return None # same contract as ``_db``: a failed open degrades to JSONL fallback def _owner_key_for_session_id(self, session_id: Optional[str]) -> Optional[str]: """The routing key that owns *session_id*, or None. The published index is authoritative; ``_session_owner_hints`` covers the window where ownership is proven but routing not yet published. Deliberately lock-free: several callers already hold ``_lock``.""" if not session_id: return None try: for entry in list(self._entries.values()): if entry.session_id == session_id: return entry.session_key except Exception: pass # bare stores / foreign entry objects in suites return (getattr(self, "_session_owner_hints", None) or {}).get(session_id) def _db_for_session_id(self, session_id: Optional[str]): """The SessionDB holding *session_id*'s row (owner from the index or a pre-published hint; unknown ids fall back to the ambient store).""" if not session_id: return self._db return self._db_for_key(self._owner_key_for_session_id(session_id)) def close_all_db_handles(self) -> None: """Close every SessionDB handle this store opened (one per path). Closing only ``store._db`` would strand secondary profiles' handles with their WAL lock held ('database is locked' on restart). Drained under the lock, closed outside it; a pinned handle is the pinner's.""" def _close(db) -> None: from hermes_state_registry import release_or_close # shared instances no-op on close() try: release_or_close(db) except Exception as exc: logger.debug("SessionDB close error during handle sweep: %s", exc) self._db_handle_cache.close_all(_close) def _ensure_loaded(self) -> None: """Load sessions index from disk if not already loaded.""" with self._lock: self._ensure_loaded_locked() def _entry_locked(self, session_key: str) -> Optional[SessionEntry]: """Load the index and return the entry for *session_key*. Lock held.""" self._ensure_loaded_locked() return self._entries.get(session_key) def _routing_scope(self) -> str: """Namespace for this store's gateway_routing rows: the resolved sessions_dir, so stores with different dirs never share entries.""" try: return str(Path(self.sessions_dir).resolve()) except Exception: return str(self.sessions_dir) def _routing_db_method(self, name: str): """Bound ``_routing_db.`` if the handle exists and has it, else None.""" method = getattr(self._routing_db or None, name, None) return method if callable(method) else None def _load_routing_rows_locked(self) -> bool: """Load state.db routing entries into ``_entries``; False when there is no loader or the load failed (warned). Lock held.""" loader = self._routing_db_method("load_gateway_routing_entries") if loader is None: return False try: for key, entry_json in loader(scope=self._routing_scope()).items(): entry = self._routing_entry_from_json(key, entry_json) if entry is not None: self._entries[key] = entry return True except Exception as e: logger.warning("gateway.session: state.db routing load failed: %s", e) return False @staticmethod def _routing_entry_from_json(key: str, entry_json: str) -> Optional[SessionEntry]: """Parse one gateway_routing row; None (with a warning) when invalid.""" from gateway.session import SessionEntry try: entry_data = json.loads(entry_json) if isinstance(entry_data, dict): return SessionEntry.from_dict(entry_data) except (ValueError, KeyError, TypeError) as e: logger.warning("Skipping invalid routing entry %r: %s", key, e) return None def _ensure_loaded_locked(self) -> None: """Load the routing index (lock held). state.db ``gateway_routing`` is primary; sessions.json is the legacy import for keys the DB lacks (persisted on the next _save). Read order (#9006 follow-up): the ``gateway_routing`` table in state.db is the primary source; sessions.json is the legacy import path for pre-migration installs (its entries are folded in for keys the DB doesn't have, then persisted to the DB on the next _save). """ if self._loaded: self._reconcile_recovered_routing_locked() return self.sessions_dir.mkdir(parents=True, exist_ok=True) db_load_succeeded = self._load_routing_rows_locked() db_had_entries = db_load_succeeded and bool(self._entries) self._import_legacy_sessions_json(db_had_entries) self._loaded = True self._routing_db_loaded = db_load_succeeded self._routing_fallback_baseline = None if db_load_succeeded else self._entries_as_dicts() # A hard crash skips graceful shutdown and leaves sessions.json pointing at ended sessions. self._prune_stale_sessions_locked() def _import_legacy_sessions_json(self, db_had_entries: bool) -> None: """Legacy import: sessions.json fills only keys the DB lacks. Lock held.""" from gateway.session import SessionEntry sessions_file = self.sessions_dir / "sessions.json" if not sessions_file.exists(): return try: with open(sessions_file, "r", encoding="utf-8-sig") as f: data = json.load(f) imported = 0 for key, entry_data in data.items(): # "_"-prefixed keys are sentinels (e.g. "_README"), not entries. if key.startswith("_") or key in self._entries: continue if not isinstance(entry_data, dict): # corrupt file must not abort the whole load logger.warning( "Skipping invalid session entry %r: expected dict, got %s", key, type(entry_data).__name__) continue try: self._entries[key] = SessionEntry.from_dict(entry_data) imported += 1 except (ValueError, KeyError, TypeError) as e: logger.warning("Skipping invalid session entry %r: %s", key, e) if imported and db_had_entries: logger.info( "gateway.session: imported %d legacy sessions.json entr%s missing from " "state.db routing table", imported, "y" if imported == 1 else "ies") except Exception as e: print(f"[gateway] Warning: Failed to load sessions: {e}") def _prune_stale_sessions_locked(self) -> None: """Remove routing entries whose session has ended in state.db (startup, lock held). Stale == ``end_reason IS NOT NULL``; rows absent from the DB are kept; a ``None`` DB handle is a no-op; DB errors are non-fatal.""" if not self._entries: return stale_keys: list = [] recovered_keys = 0 try: for key, entry in self._entries.items(): # Ask the store that owns the key, not the ambient handle, or a live # secondary-profile session gets pruned on the root copy. db = self._db_for_key(key) if db is None: continue row = db.get_session(entry.session_id) if row is None or row.get("end_reason") is None: continue verdict = self._stale_entry_verdict(key, entry, row) if verdict == "prune": stale_keys.append(key) elif verdict is not None: self._entries[key] = verdict recovered_keys += 1 except Exception as exc: logger.warning("gateway.session: stale-entry pruning skipped due to DB error: %s", exc) return for key in stale_keys: del self._entries[key] if stale_keys or recovered_keys: self._save() def _stale_entry_verdict(self, key: str, entry, row): """For a routing entry whose row has ended: ``"prune"``, a replacement entry (repoint), or None (keep as-is).""" from gateway.session_lifecycle import _now recovered_entry = None if entry.origin is not None: try: recovered_entry = self._recover_session_from_db( session_key=key, source=entry.origin, now=_now(), raise_on_lookup_error=True) except Exception as exc: # Indeterminate: keep the only routing handle. logger.debug( "gateway.session: recovery lookup failed for stale sessions.json entry %r -> " "%s: %s", key, entry.session_id, exc) return None # Compression-ended parent with a newer live child for the same peer: repoint instead of # dropping, or queued/resume-pending work vanishes until the next message. if recovered_entry is not None and recovered_entry.session_id != entry.session_id: logger.warning( "gateway.session: repointing stale sessions.json entry %r from ended %s " "(end_reason=%r) to recovered %s", key, entry.session_id, row["end_reason"], recovered_entry.session_id) if entry.prompt_pin and ( self._compression_tip_for_session_id(entry.session_id) == recovered_entry.session_id ): # The compression child continues this conversation: its internal turns keep the # pinned system bytes (the rebuilt entry is minimal and would drop them). recovered_entry.prompt_pin = entry.prompt_pin return recovered_entry # Same-id recovery == successful resume: keep the ORIGINAL entry object (the recovered one # is rebuilt minimal and would drop counters, model_override, resume markers, metadata). # A non-None recovery with the SAME session id is a successful resume (all recovery gates passed, # row reopened): keep the routing entry — it is proven valid, not a dead route (#95957). Nothing in # sessions.json changes, so no save is needed for this branch. if recovered_entry is not None: logger.info( "gateway.session: reopened ended session %s for sessions.json entry %r " "(end_reason=%r); keeping route", entry.session_id, key, row["end_reason"]) return None logger.warning( "gateway.session: pruning stale sessions.json entry %r -> %s (end_reason=%r); left by " "a crashed gateway", key, entry.session_id, row["end_reason"]) return "prune" def _entries_as_dicts(self) -> Dict[str, Any]: """Serializable snapshot of ``_entries``. Lock held.""" return {key: entry.to_dict() for key, entry in self._entries.items()} def _save(self) -> None: """Persist the routing index while the caller holds ``_lock``.""" self._persist_routing_data(*self._snapshot_routing_locked()) def _next_routing_generation_locked(self) -> int: """Bump and return the shared routing counter (lock held). Full snapshots AND single-entry fast saves MUST allocate from this one counter: the stale-write protection is a total order over serialization times and silently breaks otherwise.""" self._routing_generation = getattr(self, "_routing_generation", 0) + 1 return self._routing_generation def _reconcile_recovered_routing_locked(self) -> None: """Merge authoritative rows after a fallback-only startup load.""" baseline = getattr(self, "_routing_fallback_baseline", None) if getattr(self, "_routing_db_loaded", False) or baseline is None: return loader = self._routing_db_method("load_gateway_routing_entries") if loader is None: return try: durable = loader(scope=self._routing_scope()) except Exception as exc: logger.warning("gateway.session: recovered state.db routing load failed: %s", exc) return current = self._entries_as_dicts() for key, entry_json in durable.items(): durable_entry = self._routing_entry_from_json(key, entry_json) if durable_entry is None: continue if key not in baseline: # A key created while on fallback wins over a DB-only key; otherwise restore the # authoritative row that fallback never saw. self._entries.setdefault(key, durable_entry) elif key not in current: continue # loaded from fallback and deliberately removed elif current[key] == baseline[key]: self._entries[key] = durable_entry # unchanged fallback data yields to the DB copy self._routing_db_loaded = True self._routing_fallback_baseline = None def _snapshot_routing_locked(self) -> tuple[Dict[str, Any], int]: """Capture immutable routing data and a monotonic generation.""" self._reconcile_recovered_routing_locked() return self._entries_as_dicts(), self._next_routing_generation_locked() def _persist_routing_data(self, data: Dict[str, Any], generation: int) -> None: """Serialize all whole-index writers through one durable write lock.""" with self._lazy("_save_lock", threading.Lock): if generation <= getattr(self, "_persisted_routing_generation", 0): return # Fold in fast upserts numbered above this snapshot: they were serialized after us and # a delayed full rewrite must not regress them. fast_persisted = getattr(self, "_fast_persisted_entries", None) if fast_persisted: for key, (revision, entry_json) in fast_persisted.items(): if revision > generation: data[key] = json.loads(entry_json) db_saved = False replacer = self._routing_db_method("replace_gateway_routing_entries") if replacer is not None: try: replacer({k: json.dumps(v) for k, v in data.items()}, scope=self._routing_scope()) db_saved = True except Exception as exc: logger.warning("gateway.session: state.db routing save failed: %s", exc) if getattr(self, "_write_sessions_json", True) or not db_saved: try: self._save_sessions_json(data) except Exception as exc: if not db_saved: raise # state.db is authoritative: a failed legacy mirror must not report the # already-committed primary write as failed. logger.warning( "gateway.session: sessions.json mirror save failed after state.db commit: " "%s", exc) self._persisted_routing_generation = generation # This rewrite supersedes fast records at or below its generation; newer ones stay for # the next delayed full writer. if fast_persisted: for key in [k for k, (rev, _) in fast_persisted.items() if rev <= generation]: del fast_persisted[key] def _save_sessions_json(self, data: Dict[str, Any]) -> None: """Write the legacy sessions.json mirror of the routing index (atomic + fsync).""" atomic_json_write(self.sessions_dir / "sessions.json", {"_README": _SESSIONS_JSON_README, **data}, mode=0o600) def _save_entries(self) -> None: """Snapshot latest state under ``_lock`` and persist after releasing it.""" with self._lock: data, generation = self._snapshot_routing_locked() self._persist_routing_data(data, generation) def _save_entry( self, session_key: str, *, entry_data: Optional[Dict[str, Any]] = None, lock_held: bool = False) -> None: """Persist ONE routing entry via UPSERT — the per-turn fast path (a full rewrite fsyncs a multi-MB sessions.json). The key -> session_id mapping never changes here: structural transitions use the full rewrite (which also refreshes the sessions.json mirror; it may lag in metadata only). The revision comes from the shared routing generation counter; under ``_save_lock`` the upsert is skipped if a full snapshot or a newer fast save of this key already persisted (the reverse case lives in ``_persist_routing_data``). No DB or a failed upsert falls back to the full rewrite. ``entry_data`` persists a candidate BEFORE it is published to the live entry (failure-atomic transitions); the fallback carries it too.""" guard = contextlib.nullcontext() if lock_held else self._lock with guard: entry = self._entries.get(session_key) if entry is None: return serialized = dict(entry_data) if entry_data is not None else entry.to_dict() # The O(n) full snapshot is deferred to the fallback branch. entry_json, revision = json.dumps(serialized), self._next_routing_generation_locked() saver = self._routing_db_method("save_gateway_routing_entry") if saver is not None: try: with self._lazy("_save_lock", threading.Lock): if getattr(self, "_persisted_routing_generation", 0) >= revision: return fast_persisted = self._lazy("_fast_persisted_entries", dict) persisted = fast_persisted.get(session_key) if persisted is not None and persisted[0] >= revision: return saver(session_key, entry_json, scope=self._routing_scope()) fast_persisted[session_key] = (revision, entry_json) return except Exception as exc: logger.warning( "gateway.session: single-entry routing save failed for %r (%s); falling back " "to full index rewrite", session_key, exc) if entry_data is not None: # Full-snapshot fallback carrying the candidate transition. with guard: fallback_data = self._entries_as_dicts() fallback_data[session_key] = dict(entry_data) self._persist_routing_data(fallback_data, revision) else: self._save_entries()