refactor(gateway): move routing index to state.db, make sessions.json an optional legacy mirror (#59203)

Follow-up to #9006/#58899. The gateway routing index (session_key ->
SessionEntry) now lives in a new gateway_routing table in state.db as the
primary store; sessions.json is demoted to an optional legacy mirror.

- hermes_state.py: schema v19 — gateway_routing table (scope + session_key
  PK; scope = resolved sessions_dir so multiple stores sharing one state.db
  never cross-contaminate) with save/replace/load/delete methods
- gateway/session.py: _save() writes the whole index atomically to the DB
  (mirrors the old full-file JSON rewrite semantics) and only falls back to
  JSON when the DB write fails; _ensure_loaded reads the DB first and folds
  in legacy sessions.json entries for keys the DB lacks (pre-migration
  import; DB entries win over stale JSON)
- gateway/config.py + hermes_cli/config.py: new write_sessions_json flag
  (default true for compat/downgrade safety); gateway.write_sessions_json:
  false stops producing the file entirely
- sessions.json _README updated to say it's a legacy mirror + how to
  disable it

Rehydration is now lossless across restarts even with sessions.json deleted:
suspended/resume_pending/model_override/token state all round-trip through
the DB (the old sessions-table recovery only rebuilt the bare key mapping).

gateway/config.py

@@ -590,6 +590,13 @@ class GatewayConfig:
     
     # Storage paths
     sessions_dir: Path = field(default_factory=lambda: get_hermes_home() / "sessions")
+
+    # Whether to keep writing the legacy sessions.json mirror of the gateway
+    # routing index. The primary copy lives in state.db (gateway_routing
+    # table, #9006). Default True for backward compatibility with external
+    # tooling and downgrade safety; set gateway.write_sessions_json: false in
+    # config.yaml to stop producing the file.
+    write_sessions_json: bool = True
     
     # Delivery settings
     always_log_local: bool = True  # Always save cron outputs to local files
@@ -724,6 +731,7 @@ class GatewayConfig:
             "reset_triggers": self.reset_triggers,
             "quick_commands": self.quick_commands,
             "sessions_dir": str(self.sessions_dir),
+            "write_sessions_json": self.write_sessions_json,
             "always_log_local": self.always_log_local,
             "filter_silence_narration": self.filter_silence_narration,
             "stt_enabled": self.stt_enabled,
@@ -819,6 +827,7 @@ class GatewayConfig:
             reset_triggers=data.get("reset_triggers", ["/new", "/reset"]),
             quick_commands=quick_commands,
             sessions_dir=sessions_dir,
+            write_sessions_json=_coerce_bool(data.get("write_sessions_json"), True),
             always_log_local=_coerce_bool(data.get("always_log_local"), True),
             filter_silence_narration=_coerce_bool(
                 data.get("filter_silence_narration"), True
@@ -964,6 +973,14 @@ def load_gateway_config() -> GatewayConfig:
             if "always_log_local" in yaml_cfg:
                 gw_data["always_log_local"] = yaml_cfg["always_log_local"]
 
+            # write_sessions_json: top-level wins; nested gateway.* fallback
+            # (matches the gateway.streaming precedence pattern).
+            _gw_section = yaml_cfg.get("gateway")
+            if "write_sessions_json" in yaml_cfg:
+                gw_data["write_sessions_json"] = yaml_cfg["write_sessions_json"]
+            elif isinstance(_gw_section, dict) and "write_sessions_json" in _gw_section:
+                gw_data["write_sessions_json"] = _gw_section["write_sessions_json"]
+
             if "filter_silence_narration" in yaml_cfg:
                 gw_data["filter_silence_narration"] = yaml_cfg[
                     "filter_silence_narration"

gateway/session.py

@@ -926,6 +926,12 @@ class SessionStore:
         self._loaded = False
         self._lock = threading.Lock()
         self._has_active_processes_fn = has_active_processes_fn
+        # Whether to keep writing the legacy sessions.json mirror alongside
+        # the primary gateway_routing table in state.db. Default True for
+        # backward compatibility; disable via gateway.write_sessions_json.
+        self._write_sessions_json = bool(
+            getattr(config, "write_sessions_json", True)
+        )
         
         # Initialize SQLite session database
         self._db = None
@@ -940,24 +946,73 @@ class SessionStore:
         with self._lock:
             self._ensure_loaded_locked()
 
+    def _routing_scope(self) -> str:
+        """Namespace for this store's rows in the gateway_routing table.
+
+        The resolved sessions_dir path — the same identity that used to
+        distinguish separate sessions.json files, so two stores with
+        different directories (tests, multi-profile setups sharing one
+        state.db) never see each other's routing entries.
+        """
+        try:
+            return str(Path(self.sessions_dir).resolve())
+        except Exception:
+            return str(self.sessions_dir)
+
     def _ensure_loaded_locked(self) -> None:
-        """Load sessions index from disk. Must be called with self._lock held."""
+        """Load the routing index. Must be called with self._lock held.
+
+        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:
             return
 
         self.sessions_dir.mkdir(parents=True, exist_ok=True)
-        sessions_file = self.sessions_dir / "sessions.json"
 
+        # Primary: state.db gateway_routing table. getattr: some tests build
+        # partially-initialized stores without __init__ (same pattern as
+        # _prune_stale_sessions_locked).
+        db_had_entries = False
+        _db = getattr(self, "_db", None)
+        if _db:
+            loader = getattr(_db, "load_gateway_routing_entries", None)
+            if callable(loader):
+                try:
+                    for key, entry_json in loader(scope=self._routing_scope()).items():
+                        try:
+                            entry_data = json.loads(entry_json)
+                            if isinstance(entry_data, dict):
+                                self._entries[key] = SessionEntry.from_dict(entry_data)
+                        except (ValueError, KeyError, TypeError) as e:
+                            logger.warning(
+                                "Skipping invalid routing entry %r: %s", key, e
+                            )
+                    db_had_entries = bool(self._entries)
+                except Exception as e:
+                    logger.warning(
+                        "gateway.session: state.db routing load failed: %s", e
+                    )
+
+        # Legacy import: sessions.json (pre-migration installs, or entries
+        # written by an older gateway after a downgrade). Only fills keys the
+        # DB didn't provide — DB entries win.
+        sessions_file = self.sessions_dir / "sessions.json"
         if sessions_file.exists():
             try:
                 with open(sessions_file, "r", encoding="utf-8") as f:
                     data = json.load(f)
+                imported = 0
                 for key, entry_data in data.items():
                     # Keys starting with "_" are documentation/metadata sentinels
                     # (e.g. the "_README" note written by _save), not session
                     # entries. Skip them so they never reach SessionEntry.from_dict.
                     if key.startswith("_"):
                         continue
+                    if key in self._entries:
+                        continue
                     # Skip non-dict entries (corrupted sessions.json, e.g. a
                     # bare bool or string where a dict is expected). Without
                     # this, from_dict raises TypeError on `"origin" in data`
@@ -972,8 +1027,15 @@ class SessionStore:
                         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}")
 
@@ -1069,12 +1131,47 @@ class SessionStore:
             self._save()
 
     def _save(self) -> None:
-        """Save sessions index to disk (kept for session key -> ID mapping)."""
+        """Persist the routing index (session key -> ID mapping).
+
+        state.db's ``gateway_routing`` table is the primary store (#9006
+        follow-up): the whole index is replaced atomically in one SQLite
+        transaction, mirroring the previous full-file JSON rewrite semantics.
+
+        sessions.json is additionally written for backward compatibility
+        (external tooling, downgrade safety) unless the user disables it via
+        ``gateway.write_sessions_json: false`` in config.yaml.
+        """
+        data = {key: entry.to_dict() for key, entry in self._entries.items()}
+
+        # Primary: durable SQLite routing table.
+        db_saved = False
+        _db = getattr(self, "_db", None)
+        if _db:
+            replacer = getattr(_db, "replace_gateway_routing_entries", None)
+            if callable(replacer):
+                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
+                    )
+
+        # Legacy mirror: sessions.json. Kept on by default for compat; when
+        # disabled we still fall back to it if the DB write failed, so the
+        # index is never lost entirely.
+        if getattr(self, "_write_sessions_json", True) or not db_saved:
+            self._save_sessions_json(data)
+
+    def _save_sessions_json(self, data: Dict[str, Any]) -> None:
+        """Write the legacy sessions.json mirror of the routing index."""
         import tempfile
         self.sessions_dir.mkdir(parents=True, exist_ok=True)
         sessions_file = self.sessions_dir / "sessions.json"
 
-        data = {key: entry.to_dict() for key, entry in self._entries.items()}
         # Self-documenting sentinel so anyone who inspects this file directly
         # understands what it is and where CLI/TUI sessions actually live. Keys
         # starting with "_" are skipped on load (see _ensure_loaded_locked), so
@@ -1082,12 +1179,14 @@ class SessionStore:
         # dict so it renders at the top of the pretty-printed JSON.
         data = {
             "_README": (
-                "Gateway routing index ONLY: maps messaging session keys "
-                "(agent:main:<platform>:...) 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`. Seeing only gateway entries here is expected and "
-                "does not mean CLI sessions are missing."
+                "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:<platform>:...) 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."
             ),
             **data,
         }

hermes_cli/config.py

@@ -2760,6 +2760,13 @@ DEFAULT_CONFIG = {
         # works as a manual override and wins if set explicitly.
         "platform_connect_timeout": 30,
 
+        # Whether the gateway keeps writing the legacy sessions.json mirror of
+        # its routing index. The primary copy lives in state.db (the
+        # gateway_routing table). Default True for backward compatibility with
+        # external tooling and downgrade safety; set to false to stop
+        # producing ~/.hermes/sessions/sessions.json entirely.
+        "write_sessions_json": True,
+
         # Scale-to-zero idle detection (Phase 0). The gateway watches for idle
         # and, when an instance is opted in via the NAS "Labs" toggle (carried as
         # the HERMES_SCALE_TO_ZERO env stamp) AND messaging is relay-only/absent

hermes_state.py

@@ -122,7 +122,7 @@ T = TypeVar("T")
 
 DEFAULT_DB_PATH = get_hermes_home() / "state.db"
 
-SCHEMA_VERSION = 18
+SCHEMA_VERSION = 19
 
 # Cap on user-controlled FTS5 query input before regex/sanitizer processing.
 # Search queries do not need to be arbitrarily large, and bounding them keeps
@@ -772,6 +772,14 @@ CREATE TABLE IF NOT EXISTS state_meta (
     value TEXT
 );
 
+CREATE TABLE IF NOT EXISTS gateway_routing (
+    scope TEXT NOT NULL DEFAULT '',
+    session_key TEXT NOT NULL,
+    entry_json TEXT NOT NULL,
+    updated_at REAL NOT NULL,
+    PRIMARY KEY (scope, session_key)
+);
+
 CREATE TABLE IF NOT EXISTS compression_locks (
     session_id TEXT PRIMARY KEY,
     holder TEXT NOT NULL,
@@ -1717,6 +1725,83 @@ class SessionDB:
 
         self._execute_write(_do)
 
+    # ── Gateway routing index (replaces sessions.json, #9006 follow-up) ────
+
+    def save_gateway_routing_entry(
+        self, session_key: str, entry_json: str, *, scope: str = ""
+    ) -> None:
+        """Upsert one gateway routing entry (session_key -> SessionEntry JSON).
+
+        The gateway_routing table is the durable replacement for
+        sessions.json: one row per routing key, holding the full serialized
+        ``SessionEntry`` so the gateway can rehydrate exactly what it wrote.
+
+        ``scope`` namespaces the index the way separate sessions.json files
+        did (one per sessions_dir) — callers pass their sessions_dir path so
+        two stores with different directories never share routing state.
+        """
+        if not session_key or not entry_json:
+            return
+
+        def _do(conn):
+            conn.execute(
+                """INSERT INTO gateway_routing (scope, session_key, entry_json, updated_at)
+                   VALUES (?, ?, ?, ?)
+                   ON CONFLICT(scope, session_key) DO UPDATE SET
+                       entry_json = excluded.entry_json,
+                       updated_at = excluded.updated_at""",
+                (scope, session_key, entry_json, time.time()),
+            )
+
+        self._execute_write(_do)
+
+    def replace_gateway_routing_entries(
+        self, entries: Dict[str, str], *, scope: str = ""
+    ) -> None:
+        """Atomically replace the routing index for *scope* with *entries*.
+
+        Mirrors the sessions.json full-rewrite semantics: keys absent from
+        *entries* are removed (pruned/reset sessions disappear from the
+        index).  Runs as a single write transaction.  Other scopes are
+        untouched.
+        """
+        now = time.time()
+
+        def _do(conn):
+            conn.execute("DELETE FROM gateway_routing WHERE scope = ?", (scope,))
+            if entries:
+                conn.executemany(
+                    "INSERT INTO gateway_routing (scope, session_key, entry_json, updated_at) "
+                    "VALUES (?, ?, ?, ?)",
+                    [(scope, k, v, now) for k, v in entries.items() if k and v],
+                )
+
+        self._execute_write(_do)
+
+    def load_gateway_routing_entries(self, *, scope: str = "") -> Dict[str, str]:
+        """Load routing entries for *scope* as {session_key: entry_json}."""
+        with self._lock:
+            rows = self._conn.execute(
+                "SELECT session_key, entry_json FROM gateway_routing WHERE scope = ?",
+                (scope,),
+            ).fetchall()
+        return {r["session_key"]: r["entry_json"] for r in rows}
+
+    def delete_gateway_routing_entries(
+        self, session_keys: List[str], *, scope: str = ""
+    ) -> None:
+        """Remove routing entries for the given session keys in *scope*."""
+        if not session_keys:
+            return
+
+        def _do(conn):
+            conn.executemany(
+                "DELETE FROM gateway_routing WHERE scope = ? AND session_key = ?",
+                [(scope, k) for k in session_keys],
+            )
+
+        self._execute_write(_do)
+
     def list_gateway_sessions(
         self,
         *,

tests/gateway/test_session.py

@@ -1580,3 +1580,122 @@ class TestGatewaySessionDbRecovery:
         assert reset.session_id != entry.session_id
         assert reset.was_auto_reset is True
         assert reset.auto_reset_reason == "idle"
+
+
+class TestGatewayRoutingTable:
+    """state.db gateway_routing table is the primary routing index (#9006 follow-up)."""
+
+    @pytest.fixture(autouse=True)
+    def _isolated_db(self, tmp_path, monkeypatch):
+        # Each test gets its own state.db — DEFAULT_DB_PATH is module-level
+        # and would otherwise be shared by every SessionDB() in this file's
+        # subprocess, leaking gateway_routing rows between tests.
+        import hermes_state
+        monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", tmp_path / "state.db")
+
+    def _source(self, chat_id="chat-1", user_id="user-1"):
+        return SessionSource(
+            platform=Platform.TELEGRAM,
+            chat_id=chat_id,
+            chat_name="Alice",
+            chat_type="dm",
+            user_id=user_id,
+        )
+
+    def test_index_survives_restart_without_sessions_json(self, tmp_path):
+        """Full SessionEntry state rehydrates from state.db alone."""
+        config = GatewayConfig()
+        store = SessionStore(sessions_dir=tmp_path, config=config)
+        entry = store.get_or_create_session(self._source())
+        entry.suspended = True
+        store.set_model_override(entry.session_key, {"model": "test-model"})
+
+        # Kill the JSON mirror entirely — the DB routing table must carry
+        # the complete entry, not just the key mapping.
+        (tmp_path / "sessions.json").unlink()
+        store._db.close()
+
+        restarted = SessionStore(sessions_dir=tmp_path, config=config)
+        restarted._ensure_loaded()
+        rehydrated = restarted._entries[entry.session_key]
+        assert rehydrated.session_id == entry.session_id
+        assert rehydrated.display_name == "Alice"
+        assert rehydrated.suspended is True
+        assert rehydrated.model_override == {"model": "test-model"}
+        restarted._db.close()
+
+    def test_write_sessions_json_false_stops_producing_file(self, tmp_path):
+        config = GatewayConfig(write_sessions_json=False)
+        store = SessionStore(sessions_dir=tmp_path, config=config)
+        entry = store.get_or_create_session(self._source())
+        assert not (tmp_path / "sessions.json").exists()
+
+        # Routing still survives restart via the DB table.
+        store._db.close()
+        restarted = SessionStore(sessions_dir=tmp_path, config=config)
+        recovered = restarted.get_or_create_session(self._source())
+        assert recovered.session_id == entry.session_id
+        restarted._db.close()
+
+    def test_legacy_sessions_json_imported_when_db_table_empty(self, tmp_path):
+        """Pre-migration installs: sessions.json entries fold into the index."""
+        config = GatewayConfig()
+        store = SessionStore(sessions_dir=tmp_path, config=config)
+        entry = store.get_or_create_session(self._source())
+        store._db.close()
+
+        # Simulate a pre-migration DB: routing table empty, JSON present.
+        import hermes_state
+        db = hermes_state.SessionDB()
+        db._conn.execute("DELETE FROM gateway_routing")
+        db._conn.commit()
+        db.close()
+
+        restarted = SessionStore(sessions_dir=tmp_path, config=config)
+        recovered = restarted.get_or_create_session(self._source())
+        assert recovered.session_id == entry.session_id
+        # And the next save persists the imported entry into the DB table.
+        rows = restarted._db.load_gateway_routing_entries(
+            scope=restarted._routing_scope()
+        )
+        assert entry.session_key in rows
+        restarted._db.close()
+
+    def test_db_entries_win_over_stale_json(self, tmp_path):
+        """When both stores have a key, the DB entry is authoritative."""
+        config = GatewayConfig()
+        store = SessionStore(sessions_dir=tmp_path, config=config)
+        entry = store.get_or_create_session(self._source())
+
+        # Doctor the JSON mirror to point at a different session id.
+        data = json.loads((tmp_path / "sessions.json").read_text())
+        data[entry.session_key]["session_id"] = "20990101_000000_stale999"
+        (tmp_path / "sessions.json").write_text(json.dumps(data))
+        store._db.close()
+
+        restarted = SessionStore(sessions_dir=tmp_path, config=config)
+        restarted._ensure_loaded()
+        assert restarted._entries[entry.session_key].session_id == entry.session_id
+        restarted._db.close()
+
+    def test_prune_removes_routing_rows_for_ended_sessions(self, tmp_path):
+        """Startup prune drops ended sessions from the DB routing table too."""
+        config = GatewayConfig()
+        store = SessionStore(sessions_dir=tmp_path, config=config)
+        entry = store.get_or_create_session(self._source())
+        store._db.end_session(entry.session_id, "session_reset")
+        store._db._conn.execute(
+            "UPDATE sessions SET ended_at = 1.0, end_reason = 'session_reset' WHERE id = ?",
+            (entry.session_id,),
+        )
+        store._db._conn.commit()
+        store._db.close()
+
+        restarted = SessionStore(sessions_dir=tmp_path, config=config)
+        restarted._ensure_loaded()
+        assert entry.session_key not in restarted._entries
+        rows = restarted._db.load_gateway_routing_entries(
+            scope=restarted._routing_scope()
+        )
+        assert entry.session_key not in rows
+        restarted._db.close()