refactor: consolidate gateway session metadata into state.db

gateway/channel_directory.py

@@ -263,7 +263,67 @@ async def _build_slack(adapter) -> List[Dict[str, Any]]:
 
 
 def _build_from_sessions(platform_name: str) -> List[Dict[str, str]]:
-    """Pull known channels/contacts from sessions.json origin data."""
+    """Pull known channels/contacts from gateway session origin data.
+
+    state.db is the primary source (#9006): gateway session rows persist
+    origin_json.  Falls back to sessions.json for pre-migration databases.
+    """
+    entries = _build_from_sessions_db(platform_name)
+    if entries:
+        return entries
+    return _build_from_sessions_json(platform_name)
+
+
+def _build_from_sessions_db(platform_name: str) -> List[Dict[str, str]]:
+    """Pull channels/contacts from state.db gateway session rows."""
+    entries: List[Dict[str, str]] = []
+    try:
+        from hermes_state import SessionDB
+        db = SessionDB()
+        try:
+            lister = getattr(db, "list_gateway_sessions", None)
+            if not callable(lister):
+                return []
+            rows = lister(platform=platform_name, active_only=False)
+        finally:
+            db.close()
+
+        seen_ids = set()
+        for row in rows:
+            origin: Dict[str, Any] = {}
+            if row.get("origin_json"):
+                try:
+                    parsed = json.loads(row["origin_json"])
+                    if isinstance(parsed, dict):
+                        origin = parsed
+                except (TypeError, ValueError):
+                    pass
+            if not origin:
+                origin = {
+                    "chat_id": row.get("chat_id"),
+                    "thread_id": row.get("thread_id"),
+                    "chat_name": row.get("display_name"),
+                }
+            entry_id = _session_entry_id(origin)
+            if not entry_id or entry_id in seen_ids:
+                continue
+            seen_ids.add(entry_id)
+            entries.append({
+                "id": entry_id,
+                "name": _session_entry_name(origin),
+                "type": row.get("chat_type") or "dm",
+                "thread_id": origin.get("thread_id"),
+            })
+    except Exception as e:
+        logger.debug(
+            "Channel directory: state.db session read failed for %s: %s",
+            platform_name, e,
+        )
+    return entries
+
+
+def _build_from_sessions_json(platform_name: str) -> List[Dict[str, str]]:
+    """Legacy fallback: pull channels/contacts from sessions.json origin data."""
     sessions_path = get_hermes_home() / "sessions" / "sessions.json"
     if not sessions_path.exists():
         return []

gateway/mirror.py

@@ -102,14 +102,36 @@ def _find_session_id(
     """
     Find the active session_id for a platform + chat_id pair.
 
-    Scans sessions.json entries and matches where origin.chat_id == chat_id
-    on the right platform.  DM session keys don't embed the chat_id
-    (e.g. "agent:main:telegram:dm"), so we check the origin dict.
+    Queries state.db gateway session rows (primary source since #9006);
+    falls back to scanning sessions.json for pre-migration databases.
+    DM session keys don't embed the chat_id (e.g. "agent:main:telegram:dm"),
+    so we match on the persisted chat origin, not the key.
 
     When *user_id* is provided, prefer exact sender matches. If multiple
     same-chat candidates exist and none matches the user, return None instead
     of guessing and contaminating another participant's session.
     """
+    # Primary: state.db
+    try:
+        from hermes_state import SessionDB
+        db = SessionDB()
+        try:
+            finder = getattr(db, "find_session_by_origin", None)
+            if callable(finder):
+                session_id = finder(
+                    platform=platform,
+                    chat_id=chat_id,
+                    thread_id=thread_id,
+                    user_id=user_id,
+                )
+                if session_id:
+                    return str(session_id)
+        finally:
+            db.close()
+    except Exception as e:
+        logger.debug("Mirror state.db session lookup failed: %s", e)
+
+    # Fallback: sessions.json (pre-migration databases)
     if not _SESSIONS_INDEX.exists():
         return None
 

gateway/run.py

@@ -7540,14 +7540,11 @@ async def _session_expiry_watcher(self, interval: int = 300):
                         _update_prompt_pending = getattr(self, "_update_prompt_pending", None)
                         if isinstance(_update_prompt_pending, dict):
                             _update_prompt_pending.pop(key, None)
-                        with self.session_store._lock:
-                            entry.expiry_finalized = True
-                            # Session finalization is a conversation boundary —
-                            # drop the persisted /model override too so a later
-                            # message doesn't rehydrate it after the in-memory
-                            # override was popped above.
-                            entry.model_override = None
-                            self.session_store._save()
+                        # Persist the finalized flag to sessions.json AND
+                        # state.db (single write-path, #9006) — also drops
+                        # the persisted /model override, since finalization
+                        # is a conversation boundary.
+                        self.session_store.set_expiry_finalized(entry)
                         logger.debug(
                             "Session expiry finalized for %s",
                             entry.session_id,
@@ -7562,9 +7559,9 @@ async def _session_expiry_watcher(self, interval: int = 300):
                                 "Marking as finalized to prevent infinite retry loop.",
                                 failures, entry.session_id, e,
                             )
-                            with self.session_store._lock:
-                                entry.expiry_finalized = True
-                                self.session_store._save()
+                            self.session_store.set_expiry_finalized(
+                                entry, clear_model_override=False
+                            )
                             _finalize_failures.pop(entry.session_id, None)
                         else:
                             logger.debug(

gateway/session.py

@@ -1202,6 +1202,7 @@ def _record_gateway_session_peer(
         session_id: str,
         session_key: str,
         source: Optional[SessionSource],
+        display_name: Optional[str] = None,
     ) -> None:
         """Persist the routing peer for an existing gateway session row."""
         if not self._db or not source:
@@ -1210,6 +1211,11 @@ def _record_gateway_session_peer(
         if not callable(recorder):
             return
         try:
+            origin_json = None
+            try:
+                origin_json = json.dumps(source.to_dict())
+            except Exception:
+                pass
             recorder(
                 session_id,
                 source=source.platform.value,
@@ -1218,9 +1224,56 @@ def _record_gateway_session_peer(
                 chat_id=source.chat_id,
                 chat_type=source.chat_type,
                 thread_id=source.thread_id,
+                display_name=display_name or source.chat_name,
+                origin_json=origin_json,
             )
+        except TypeError:
+            # Older SessionDB without display_name/origin_json kwargs.
+            try:
+                recorder(
+                    session_id,
+                    source=source.platform.value,
+                    user_id=source.user_id,
+                    session_key=session_key,
+                    chat_id=source.chat_id,
+                    chat_type=source.chat_type,
+                    thread_id=source.thread_id,
+                )
+            except Exception as exc:
+                logger.debug("Gateway session peer record failed for %s: %s", session_key, exc)
         except Exception as exc:
             logger.debug("Gateway session peer record failed for %s: %s", session_key, exc)
+
+    def set_expiry_finalized(
+        self, entry: SessionEntry, *, clear_model_override: bool = True
+    ) -> None:
+        """Mark a session entry expiry-finalized in memory, sessions.json, AND state.db.
+
+        Single write-path for the expiry watcher (#9006): keeps the durable
+        state.db flag in sync with the JSON routing index so the flag
+        survives sessions.json pruning/loss.
+
+        ``clear_model_override=False`` preserves the give-up path's original
+        behavior (flag only, no override drop).
+        """
+        with self._lock:
+            entry.expiry_finalized = True
+            if clear_model_override:
+                # Session finalization is a conversation boundary — drop the
+                # persisted /model override too so a later message doesn't
+                # rehydrate it after the in-memory override was popped.
+                entry.model_override = None
+            self._save()
+        if self._db:
+            setter = getattr(self._db, "set_expiry_finalized", None)
+            if callable(setter):
+                try:
+                    setter(entry.session_id, True)
+                except Exception as exc:
+                    logger.debug(
+                        "Session DB expiry_finalized write failed for %s: %s",
+                        entry.session_id, exc,
+                    )
 
     def _is_session_expired(self, entry: SessionEntry) -> bool:
         """Check if a session has expired based on its reset policy.
@@ -1618,6 +1671,7 @@ def get_or_create_session(
                     session_id,
                     session_key,
                     source,
+                    display_name=entry.display_name,
                 )
             except Exception as e:
                 print(f"[gateway] Warning: Failed to create SQLite session: {e}")
@@ -1643,6 +1697,7 @@ def update_session(
                     entry.session_id,
                     session_key,
                     entry.origin,
+                    display_name=entry.display_name,
                 )
 
     def set_model_override(
@@ -1888,6 +1943,7 @@ def reset_session(self, session_key: str, display_name: Optional[str] = None) ->
                     session_id,
                     session_key,
                     old_entry.origin,
+                    display_name=new_entry.display_name if new_entry else None,
                 )
             except Exception as e:
                 logger.debug("Session DB operation failed: %s", e)
@@ -1950,6 +2006,7 @@ def switch_session(self, session_key: str, target_session_id: str) -> Optional[S
                 target_session_id,
                 session_key,
                 new_entry.origin if new_entry else None,
+                display_name=new_entry.display_name if new_entry else None,
             )
 
         return new_entry

hermes_cli/status.py

@@ -543,17 +543,39 @@ def _resolve_env(env_ref) -> str:
     print()
     print(color("◆ Sessions", Colors.CYAN, Colors.BOLD))
 
-    sessions_file = get_hermes_home() / "sessions" / "sessions.json"
-    if sessions_file.exists():
-        import json
+    # Gateway session count: state.db is the source of truth (#9006);
+    # fall back to sessions.json for pre-migration installs.
+    _session_count = None
+    try:
+        from hermes_state import SessionDB
+        _db = SessionDB()
         try:
-            with open(sessions_file, encoding="utf-8") as f:
-                data = json.load(f)
-                print(f"  Active:       {len(data)} session(s)")
-        except Exception:
-            print("  Active:       (error reading sessions file)")
+            _lister = getattr(_db, "list_gateway_sessions", None)
+            if callable(_lister):
+                _session_count = len(_lister(active_only=True))
+        finally:
+            _db.close()
+    except Exception:
+        _session_count = None
+
+    if _session_count is not None and _session_count > 0:
+        print(f"  Active:       {_session_count} session(s)")
     else:
-        print("  Active:       0")
+        sessions_file = get_hermes_home() / "sessions" / "sessions.json"
+        if sessions_file.exists():
+            import json
+            try:
+                with open(sessions_file, encoding="utf-8") as f:
+                    data = json.load(f)
+                    _entries = {
+                        k: v for k, v in data.items()
+                        if not str(k).startswith("_")
+                    } if isinstance(data, dict) else {}
+                    print(f"  Active:       {len(_entries)} session(s)")
+            except Exception:
+                print("  Active:       (error reading sessions file)")
+        else:
+            print(f"  Active:       {_session_count if _session_count is not None else 0}")
 
     # =========================================================================
     # Deep checks