refactor: consolidate gateway session metadata into state.db (#58899)

Moves gateway routing metadata (display_name, origin_json, expiry_finalized)
into state.db, making SQLite the single source of truth for gateway session
discovery. Eliminates the dual-file (sessions.json + state.db) polling
dependency that caused the mcp_serve new-conversation race (#8925).

- hermes_state.py: schema v18 (3 new sessions columns + sessions.json
  backfill migration), record_gateway_session_peer gains
  display_name/origin_json, new set_expiry_finalized(),
  list_gateway_sessions(), find_session_by_origin()
- gateway/session.py: peer recorder persists display_name + full origin
  JSON; new SessionStore.set_expiry_finalized() single write-path
- gateway/run.py: expiry watcher success + give-up paths use the store
  helper so the flag lands in both sessions.json and state.db
- mcp_serve.py: routing index reads state.db first (sessions.json fallback
  for pre-migration DBs); _poll_once collapses to a single state.db mtime
  check — the #8925 race is structurally impossible now
- gateway/mirror.py, gateway/channel_directory.py, hermes_cli/status.py:
  query state.db first, sessions.json fallback

Closes #9006

mcp_serve.py

@@ -37,6 +37,7 @@ import sys
 import threading
 import time
 from dataclasses import dataclass, field
+from datetime import datetime
 from pathlib import Path
 from typing import Dict, List, Optional
 
@@ -79,10 +80,99 @@ def _get_session_db():
 
 
 def _load_sessions_index() -> dict:
-    """Load the gateway sessions.json index directly.
+    """Load the gateway session routing index.
 
     Returns a dict of session_key -> entry_dict with platform routing info.
-    This avoids importing the full SessionStore which needs GatewayConfig.
+
+    state.db is the primary source (#9006): gateway sessions persist their
+    routing metadata (session_key, chat/thread ids, display_name, origin) on
+    the durable session row, so a single database read replaces the old
+    dual-file sessions.json dependency.  Falls back to sessions.json for
+    pre-migration databases where no gateway rows carry a session_key yet.
+    """
+    entries = _load_sessions_index_from_db()
+    if entries:
+        return entries
+    return _load_sessions_index_from_json()
+
+
+def _row_to_index_entry(row: dict) -> dict:
+    """Convert a state.db gateway session row to the sessions.json entry shape."""
+    origin = {}
+    origin_json = row.get("origin_json")
+    if origin_json:
+        try:
+            parsed = json.loads(origin_json)
+            if isinstance(parsed, dict):
+                origin = parsed
+        except (TypeError, ValueError):
+            pass
+    if not origin:
+        # Pre-origin_json rows: synthesize the minimal origin from columns.
+        origin = {
+            "platform": row.get("source", ""),
+            "chat_id": row.get("chat_id"),
+            "chat_type": row.get("chat_type"),
+            "thread_id": row.get("thread_id"),
+            "user_id": row.get("user_id"),
+        }
+
+    def _iso(ts) -> str:
+        try:
+            return datetime.fromtimestamp(float(ts)).isoformat() if ts else ""
+        except (TypeError, ValueError, OSError):
+            return ""
+
+    input_tokens = int(row.get("input_tokens") or 0)
+    output_tokens = int(row.get("output_tokens") or 0)
+    return {
+        "session_id": str(row.get("id", "")),
+        "session_key": row.get("session_key", ""),
+        "platform": row.get("source", ""),
+        "chat_type": row.get("chat_type") or origin.get("chat_type", ""),
+        "display_name": row.get("display_name") or origin.get("chat_name") or "",
+        "origin": origin,
+        "created_at": _iso(row.get("started_at")),
+        "updated_at": _iso(row.get("last_active") or row.get("started_at")),
+        "input_tokens": input_tokens,
+        "output_tokens": output_tokens,
+        "total_tokens": input_tokens + output_tokens,
+    }
+
+
+def _load_sessions_index_from_db() -> dict:
+    """Build the routing index from state.db gateway session rows."""
+    db = _get_session_db()
+    if db is None:
+        return {}
+    try:
+        lister = getattr(db, "list_gateway_sessions", None)
+        if not callable(lister):
+            return {}
+        rows = lister(active_only=True)
+        entries = {}
+        for row in rows:
+            key = row.get("session_key")
+            if not key:
+                continue
+            entries[key] = _row_to_index_entry(row)
+        return entries
+    except Exception as e:
+        logger.debug("Failed to load gateway sessions from state.db: %s", e)
+        return {}
+    finally:
+        try:
+            db.close()
+        except Exception:
+            pass
+
+
+def _load_sessions_index_from_json() -> dict:
+    """Legacy fallback: load the gateway sessions.json index directly.
+
+    Used only for pre-migration databases whose gateway rows don't carry a
+    session_key yet.  This avoids importing the full SessionStore which
+    needs GatewayConfig.
     """
     sessions_file = _get_sessions_dir() / "sessions.json"
     if not sessions_file.exists():
@@ -226,8 +316,7 @@ class EventBridge:
         self._last_poll_timestamps: Dict[str, float] = {}  # session_key -> unix timestamp
         # In-memory approval tracking (populated from events)
         self._pending_approvals: Dict[str, dict] = {}
-        # mtime cache — skip expensive work when files haven't changed
-        self._sessions_json_mtime: float = 0.0
+        # mtime cache — skip expensive work when state.db hasn't changed
         self._state_db_mtime: float = 0.0
         self._cached_sessions_index: dict = {}
 
@@ -353,24 +442,14 @@ class EventBridge:
     def _poll_once(self, db):
         """Check for new messages across all sessions.
 
-        Uses mtime checks on sessions.json and state.db to skip work
-        when nothing has changed — makes 200ms polling essentially free.
+        Uses a single mtime check on state.db to skip work when nothing
+        has changed — makes 200ms polling essentially free.  Since #9006
+        the routing index itself lives in state.db (session rows carry
+        session_key/origin metadata), so a new conversation and its first
+        message land in the SAME file and one mtime check covers both —
+        eliminating the old dual-file (sessions.json + state.db) race that
+        could drop brand-new conversations (#8925).
         """
-        # Check if sessions.json has changed (mtime check is ~1μs).
-        # Capture the previously-seen mtime *before* refreshing the cache so the
-        # skip guard below can still tell whether sessions.json changed this tick.
-        prev_sessions_json_mtime = self._sessions_json_mtime
-        sessions_file = _get_sessions_dir() / "sessions.json"
-        try:
-            sj_mtime = sessions_file.stat().st_mtime if sessions_file.exists() else 0.0
-        except OSError:
-            sj_mtime = 0.0
-
-        if sj_mtime != self._sessions_json_mtime:
-            self._sessions_json_mtime = sj_mtime
-            self._cached_sessions_index = _load_sessions_index()
-
-        # Check if state.db has changed
         try:
             from hermes_constants import get_hermes_home
             db_file = get_hermes_home() / "state.db"
@@ -382,19 +461,14 @@ class EventBridge:
         except OSError:
             db_mtime = 0.0
 
-        # Skip only when NEITHER file changed since the last poll. Comparing
-        # against ``prev_sessions_json_mtime`` (not the freshly-stored
-        # ``self._sessions_json_mtime``) is essential: the latter was just set to
-        # ``sj_mtime`` above, so using it would make the sessions.json term
-        # always true and collapse the guard to a db-only check. That would
-        # discard a tick where only sessions.json changed — e.g. a brand-new
-        # conversation registered after its first message already landed in
-        # state.db on an earlier tick — and the new chat's messages would never
-        # be emitted until state.db happened to change again.
-        if db_mtime == self._state_db_mtime and sj_mtime == prev_sessions_json_mtime:
+        if db_mtime == self._state_db_mtime:
             return  # Nothing changed since last poll — skip entirely
 
         self._state_db_mtime = db_mtime
+        # Refresh the routing index from state.db on every change tick —
+        # it's a single indexed query and it can never lag the messages
+        # table (both live in the same database file).
+        self._cached_sessions_index = _load_sessions_index()
         entries = self._cached_sessions_index
 
         for session_key, entry in entries.items():