Persisting gateway session chat metadata in state.db

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

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

  merged to main

  files: gateway/channel_directory.py, gateway/mirror.py, gateway/run.py, gateway/session.py, hermes_cli/status.py, hermes_state.py, mcp_serve.py, tests/test_hermes_state.py, tests/test_mcp_serve.py

  Show the diff

  • 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

  • hermes_state.py

    @@ -122,7 +122,7 @@ def _delete_delegate_children(conn, parent_ids: List[str]) -> List[str]:
     
     DEFAULT_DB_PATH = get_hermes_home() / "state.db"
     
    -SCHEMA_VERSION = 17
    +SCHEMA_VERSION = 18
     
     # Cap on user-controlled FTS5 query input before regex/sanitizer processing.
     # Search queries do not need to be arbitrarily large, and bounding them keeps
    @@ -705,6 +705,9 @@ def repair_state_db_schema(db_path: Path, *, backup: bool = True) -> Dict[str, A
         chat_id TEXT,
         chat_type TEXT,
         thread_id TEXT,
    +    display_name TEXT,
    +    origin_json TEXT,
    +    expiry_finalized INTEGER DEFAULT 0,
         model TEXT,
         model_config TEXT,
         system_prompt TEXT,
    @@ -1522,6 +1525,19 @@ def _init_schema(self):
                         )
                     except sqlite3.OperationalError:
                         pass
    +            if current_version < 18:
    +                # v18: gateway metadata consolidation (#9006). Backfill
    +                # display_name / origin_json / expiry_finalized from
    +                # sessions.json so pre-migration gateway sessions are
    +                # discoverable from state.db without the JSON index.
    +                try:
    +                    self._backfill_gateway_metadata_from_sessions_json(cursor)
    +                except Exception as exc:
    +                    # Backfill is best-effort: sessions.json may be absent,
    +                    # corrupted, or partially stale. Missing metadata simply
    +                    # means consumers fall back to sessions.json for those
    +                    # rows until the gateway rewrites them.
    +                    logger.debug("v18 gateway metadata backfill skipped: %s", exc)
                 if current_version < SCHEMA_VERSION and fts_migrations_complete:
                     cursor.execute(
                         "UPDATE schema_version SET version = ?",
    @@ -1647,8 +1663,17 @@ def record_gateway_session_peer(
             chat_id: str = None,
             chat_type: str = None,
             thread_id: str = None,
    +        display_name: str = None,
    +        origin_json: str = None,
         ) -> None:
    -        """Persist the gateway routing peer for an existing session row."""
    +        """Persist the gateway routing peer for an existing session row.
    +
    +        ``display_name`` / ``origin_json`` carry the gateway's presentation
    +        and full origin metadata (#9006) so consumers (mcp_serve, mirror,
    +        channel directory) can read routing data from state.db instead of
    +        sessions.json.  They are COALESCE'd only in the sense that ``None``
    +        leaves the existing value untouched.
    +        """
             if not session_id or not session_key:
                 return
     
    @@ -1656,7 +1681,9 @@ def _do(conn):
                 conn.execute(
                     """UPDATE sessions
                        SET session_key = ?, source = ?, user_id = ?, chat_id = ?,
    -                       chat_type = ?, thread_id = ?
    +                       chat_type = ?, thread_id = ?,
    +                       display_name = COALESCE(?, display_name),
    +                       origin_json = COALESCE(?, origin_json)
                        WHERE id = ?""",
                     (
                         session_key,
    @@ -1665,12 +1692,168 @@ def _do(conn):
                         chat_id,
                         chat_type,
                         thread_id,
    +                    display_name,
    +                    origin_json,
                         session_id,
                     ),
                 )
     
             self._execute_write(_do)
     
    +    def set_expiry_finalized(self, session_id: str, finalized: bool = True) -> None:
    +        """Mark a gateway session's expiry-finalization flag in state.db.
    +
    +        Mirrors ``SessionEntry.expiry_finalized`` (sessions.json) so the flag
    +        survives even if the JSON index is pruned or lost (#9006).
    +        """
    +        if not session_id:
    +            return
    +
    +        def _do(conn):
    +            conn.execute(
    +                "UPDATE sessions SET expiry_finalized = ? WHERE id = ?",
    +                (1 if finalized else 0, session_id),
    +            )
    +
    +        self._execute_write(_do)
    +
    +    def list_gateway_sessions(
    +        self,
    +        *,
    +        platform: Optional[str] = None,
    +        active_only: bool = True,
    +    ) -> List[Dict[str, Any]]:
    +        """List gateway sessions (rows with a session_key) from state.db.
    +
    +        Returns the newest row per session_key — the same shape consumers got
    +        from sessions.json: one live mapping per routing key.  ``platform``
    +        filters on ``source``; ``active_only`` restricts to sessions that
    +        have not ended.
    +        """
    +        query = """
    +            SELECT sessions.*,
    +                   COALESCE(
    +                       (SELECT MAX(m.timestamp) FROM messages m
    +                        WHERE m.session_id = sessions.id),
    +                       sessions.started_at
    +                   ) AS last_active
    +            FROM sessions
    +            WHERE session_key IS NOT NULL
    +              AND started_at = (
    +                  SELECT MAX(s2.started_at) FROM sessions s2
    +                  WHERE s2.session_key = sessions.session_key
    +              )
    +        """
    +        params: list = []
    +        if platform:
    +            query += " AND LOWER(source) = LOWER(?)"
    +            params.append(platform)
    +        if active_only:
    +            query += " AND ended_at IS NULL"
    +        query += " ORDER BY last_active DESC"
    +        with self._lock:
    +            rows = self._conn.execute(query, params).fetchall()
    +        return [dict(r) for r in rows]
    +
    +    def find_session_by_origin(
    +        self,
    +        *,
    +        platform: str,
    +        chat_id: str,
    +        thread_id: Optional[str] = None,
    +        user_id: Optional[str] = None,
    +    ) -> Optional[str]:
    +        """Find the most recent live session_id for a platform + chat origin.
    +
    +        Equivalent of gateway/mirror's sessions.json scan: matches on
    +        source + chat_id (+ thread_id when provided).  When ``user_id`` is
    +        provided, exact sender matches are preferred; if multiple distinct
    +        users share the chat and none matches, returns None rather than
    +        contaminating another participant's session.
    +        """
    +        if not platform or chat_id in (None, ""):
    +            return None
    +        query = """
    +            SELECT id, user_id, started_at FROM sessions
    +            WHERE LOWER(source) = LOWER(?)
    +              AND session_key IS NOT NULL
    +              AND chat_id = ?
    +              AND ended_at IS NULL
    +        """
    +        params: list = [platform, str(chat_id)]
    +        if thread_id is not None:
    +            query += " AND COALESCE(thread_id, '') = ?"
    +            params.append(str(thread_id))
    +        query += " ORDER BY started_at DESC"
    +        with self._lock:
    +            rows = [dict(r) for r in self._conn.execute(query, params).fetchall()]
    +        if not rows:
    +            return None
    +        if user_id:
    +            exact = [r for r in rows if str(r.get("user_id") or "") == str(user_id)]
    +            if exact:
    +                return str(exact[0]["id"])
    +            if len(rows) > 1:
    +                return None
    +        elif len(rows) > 1:
    +            distinct_users = {
    +                str(r.get("user_id") or "").strip()
    +                for r in rows
    +                if str(r.get("user_id") or "").strip()
    +            }
    +            if len(distinct_users) > 1:
    +                return None
    +        return str(rows[0]["id"])
    +
    +    def _backfill_gateway_metadata_from_sessions_json(
    +        self, cursor: sqlite3.Cursor
    +    ) -> None:
    +        """One-time v18 backfill of gateway metadata from sessions.json.
    +
    +        Existing gateway sessions predate the display_name / origin_json /
    +        expiry_finalized columns; copy what sessions.json knows so consumers
    +        can switch to state.db without losing pre-migration sessions.
    +        Only fills NULL columns — never overwrites data written by newer code.
    +        """
    +        sessions_file = get_hermes_home() / "sessions" / "sessions.json"
    +        if not sessions_file.exists():
    +            return
    +        with open(sessions_file, "r", encoding="utf-8") as f:
    +            data = json.load(f)
    +        if not isinstance(data, dict):
    +            return
    +        for key, entry in data.items():
    +            if str(key).startswith("_") or not isinstance(entry, dict):
    +                continue
    +            session_id = entry.get("session_id")
    +            if not session_id:
    +                continue
    +            origin = entry.get("origin")
    +            cursor.execute(
    +                """UPDATE sessions
    +                   SET session_key = COALESCE(session_key, ?),
    +                       chat_id = COALESCE(chat_id, ?),
    +                       chat_type = COALESCE(chat_type, ?),
    +                       thread_id = COALESCE(thread_id, ?),
    +                       display_name = COALESCE(display_name, ?),
    +                       origin_json = COALESCE(origin_json, ?),
    +                       expiry_finalized = CASE
    +                           WHEN COALESCE(expiry_finalized, 0) = 0 AND ? = 1 THEN 1
    +                           ELSE expiry_finalized
    +                       END
    +                   WHERE id = ?""",
    +                (
    +                    entry.get("session_key") or key,
    +                    (origin or {}).get("chat_id") if isinstance(origin, dict) else None,
    +                    entry.get("chat_type"),
    +                    (origin or {}).get("thread_id") if isinstance(origin, dict) else None,
    +                    entry.get("display_name"),
    +                    json.dumps(origin) if isinstance(origin, dict) else None,
    +                    1 if entry.get("expiry_finalized") or entry.get("memory_flushed") else 0,
    +                    str(session_id),
    +                ),
    +            )
    +
         def find_latest_gateway_session_for_peer(
             self,
             *,

  • mcp_serve.py

    @@ -37,6 +37,7 @@
     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 = [redacted:secret-assignment]"input_tokens") or 0)
    +    output_tokens = [redacted:secret-assignment]"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": [redacted:secret-assignment]
    +        "output_tokens": [redacted:secret-assignment]
    +        "total_tokens": [redacted:secret-assignment] + 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 @@ def __init__(self):
             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 @@ def _poll_loop(self):
         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 @@ def _poll_once(self, db):
             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():

  • tests/test_hermes_state.py

    @@ -2,6 +2,7 @@
     
     import sqlite3
     import time
    +import json
     import pytest
     
     import hermes_state
    @@ -4831,6 +4832,174 @@ def test_gateway_session_recovery_reopens_legacy_agent_close_rows(db):
         ) is None
     
     
    +def test_gateway_metadata_display_name_origin_round_trip(db):
    +    """record_gateway_session_peer persists display_name/origin_json (#9006)."""
    +    db.create_session("gw-meta", "telegram", user_id="u1")
    +    origin = {"platform": "telegram", "chat_id": "c1", "chat_name": "Alice", "chat_type": "dm"}
    +    db.record_gateway_session_peer(
    +        "gw-meta",
    +        source="telegram",
    +        user_id="u1",
    +        session_key="agent:main:telegram:dm:c1",
    +        chat_id="c1",
    +        chat_type="dm",
    +        thread_id=None,
    +        display_name="Alice",
    +        origin_json=json.dumps(origin),
    +    )
    +    row = db.get_session("gw-meta")
    +    assert row["display_name"] == "Alice"
    +    assert json.loads(row["origin_json"])["chat_name"] == "Alice"
    +
    +    # None values must not clobber existing metadata.
    +    db.record_gateway_session_peer(
    +        "gw-meta",
    +        source="telegram",
    +        user_id="u1",
    +        session_key="agent:main:telegram:dm:c1",
    +        chat_id="c1",
    +        chat_type="dm",
    +    )
    +    row = db.get_session("gw-meta")
    +    assert row["display_name"] == "Alice"
    +    assert row["origin_json"] is not None
    +
    +
    +def test_set_expiry_finalized_round_trip(db):
    +    db.create_session("gw-exp", "telegram", session_key="agent:main:telegram:dm:x")
    +    row = db.get_session("gw-exp")
    +    assert not row["expiry_finalized"]
    +    db.set_expiry_finalized("gw-exp")
    +    assert db.get_session("gw-exp")["expiry_finalized"] == 1
    +    db.set_expiry_finalized("gw-exp", False)
    +    assert db.get_session("gw-exp")["expiry_finalized"] == 0
    +
    +
    +def test_list_gateway_sessions_filters_and_dedupes(db):
    +    # Two rows on the same session_key: only the newest should be returned.
    +    db.create_session(
    +        "gw-old", "telegram",
    +        session_key="agent:main:telegram:dm:c1", chat_id="c1", chat_type="dm",
    +    )
    +    db._conn.execute(
    +        "UPDATE sessions SET started_at = started_at - 100 WHERE id = 'gw-old'"
    +    )
    +    db._conn.commit()
    +    db.create_session(
    +        "gw-new", "telegram",
    +        session_key="agent:main:telegram:dm:c1", chat_id="c1", chat_type="dm",
    +    )
    +    db.create_session(
    +        "gw-discord", "discord",
    +        session_key="agent:main:discord:group:g1:u1", chat_id="g1", chat_type="group",
    +    )
    +    # Non-gateway session (no session_key) must never appear.
    +    db.create_session("cli-session", "cli")
    +    # Ended gateway session excluded when active_only.
    +    db.create_session(
    +        "gw-ended", "slack",
    +        session_key="agent:main:slack:dm:s1", chat_id="s1", chat_type="dm",
    +    )
    +    db.end_session("gw-ended", "session_reset")
    +
    +    rows = db.list_gateway_sessions(active_only=True)
    +    ids = {r["id"] for r in rows}
    +    assert ids == {"gw-new", "gw-discord"}
    +
    +    tg_rows = db.list_gateway_sessions(platform="telegram", active_only=True)
    +    assert [r["id"] for r in tg_rows] == ["gw-new"]
    +
    +    all_rows = db.list_gateway_sessions(active_only=False)
    +    assert "gw-ended" in {r["id"] for r in all_rows}
    +    assert "cli-session" not in {r["id"] for r in all_rows}
    +
    +
    +def test_find_session_by_origin_matching_rules(db):
    +    db.create_session(
    +        "gw-o1", "telegram", user_id="u1",
    +        session_key="agent:main:telegram:group:c9:u1", chat_id="c9", chat_type="group",
    +    )
    +    db.create_session(
    +        "gw-o2", "telegram", user_id="u2",
    +        session_key="agent:main:telegram:group:c9:u2", chat_id="c9", chat_type="group",
    +    )
    +
    +    # Exact user match wins.
    +    assert db.find_session_by_origin(
    +        platform="telegram", chat_id="c9", user_id="u2"
    +    ) == "gw-o2"
    +    # Unknown user among multiple distinct users -> None (no contamination).
    +    assert db.find_session_by_origin(
    +        platform="telegram", chat_id="c9", user_id="u3"
    +    ) is None
    +    # No user given + multiple distinct users -> None.
    +    assert db.find_session_by_origin(platform="telegram", chat_id="c9") is None
    +    # Ended sessions are ignored: only gw-o1 remains as a live candidate.
    +    # A single remaining candidate is returned even without an exact user
    +    # match — mirrors the original sessions.json scan semantics.
    +    db.end_session("gw-o2", "session_reset")
    +    assert db.find_session_by_origin(
    +        platform="telegram", chat_id="c9", user_id="u2"
    +    ) == "gw-o1"
    +    # Single remaining candidate resolves without user_id.
    +    assert db.find_session_by_origin(platform="telegram", chat_id="c9") == "gw-o1"
    +    # Thread filter.
    +    db.create_session(
    +        "gw-th", "discord", user_id="u9",
    +        session_key="agent:main:discord:thread:t7", chat_id="ch7",
    +        chat_type="thread", thread_id="t7",
    +    )
    +    assert db.find_session_by_origin(
    +        platform="discord", chat_id="ch7", thread_id="t7"
    +    ) == "gw-th"
    +    assert db.find_session_by_origin(
    +        platform="discord", chat_id="ch7", thread_id="other"
    +    ) is None
    +
    +
    +def test_v18_backfill_from_sessions_json(tmp_path, monkeypatch):
    +    """Migration backfills display_name/origin_json/expiry_finalized from sessions.json."""
    +    import hermes_state as hs
    +
    +    home = tmp_path / ".hermes"
    +    (home / "sessions").mkdir(parents=True)
    +    monkeypatch.setenv("HERMES_HOME", str(home))
    +    monkeypatch.setattr(hs, "DEFAULT_DB_PATH", home / "state.db")
    +
    +    # Seed a pre-v18 database: create schema, downgrade version, add a bare row.
    +    db = hs.SessionDB(home / "state.db")
    +    db.create_session("legacy-gw", "telegram", user_id="u1")
    +    db._conn.execute("UPDATE schema_version SET version = 17")
    +    db._conn.execute(
    +        "UPDATE sessions SET session_key = NULL, display_name = NULL, "
    +        "origin_json = NULL WHERE id = 'legacy-gw'"
    +    )
    +    db._conn.commit()
    +    db.close()
    +
    +    origin = {"platform": "telegram", "chat_id": "123", "chat_name": "Alice",
    +              "chat_type": "dm", "user_id": "u1"}
    +    (home / "sessions" / "sessions.json").write_text(json.dumps({
    +        "_README": "sentinel",
    +        "agent:main:telegram:dm:123": {
    +            "session_id": "legacy-gw",
    +            "display_name": "Alice",
    +            "chat_type": "dm",
    +            "expiry_finalized": True,
    +            "origin": origin,
    +        },
    +    }))
    +
    +    db = hs.SessionDB(home / "state.db")
    +    row = db.get_session("legacy-gw")
    +    db.close()
    +    assert row["session_key"] == "agent:main:telegram:dm:123"
    +    assert row["display_name"] == "Alice"
    +    assert row["chat_id"] == "123"
    +    assert json.loads(row["origin_json"])["chat_name"] == "Alice"
    +    assert row["expiry_finalized"] == 1
    +
    +
     def test_compression_failure_cooldown_round_trips_and_clears(db):
         db.create_session("s1", "cli")
     

  • tests/test_mcp_serve.py

    @@ -1232,18 +1232,19 @@ def get_messages(self, sid):
             assert len(r2["events"]) == 1
             assert r2["events"][0]["content"] == "New reply!"
     
    -    def test_poll_picks_up_new_conversation_when_only_sessions_json_changed(
    +    def test_poll_picks_up_new_conversation_on_db_change(
             self, tmp_path, monkeypatch
         ):
    -        """A conversation registered in sessions.json *after* its first message
    -        already landed in state.db must still be picked up, even when state.db's
    -        mtime did not move on this tick.
    -
    -        Regression guard: the skip-when-unchanged check must compare against the
    -        sessions.json mtime seen on the *previous* poll. If it instead compares
    -        against the just-refreshed value, the sessions.json term is always true,
    -        the guard collapses to a db-only check, and a sessions.json-only change
    -        is silently dropped — the new chat's messages never reach MCP clients.
    +        """A brand-new conversation must be picked up on the tick where
    +        state.db changes.
    +
    +        Since #9006 the routing index lives IN state.db (session rows carry
    +        session_key/origin metadata), so a new conversation's registration and
    +        its first message land in the same file — a single mtime check covers
    +        both and the old dual-file (sessions.json + state.db) race (#8925) is
    +        structurally impossible. This test asserts the index is refreshed on a
    +        db-mtime bump, so a conversation the bridge has never seen before is
    +        emitted on the same tick.
             """
             import mcp_serve
     
    @@ -1257,14 +1258,20 @@ def test_poll_picks_up_new_conversation_when_only_sessions_json_changed(
             db_path.write_text("placeholder")
     
             session_id = "20260329_150000_late_register"
    -        sessions_json = sessions_dir / "sessions.json"
    -        sessions_json.write_text(json.dumps({
    -            "agent:main:telegram:dm:late": {
    -                "session_id": session_id,
    -                "platform": "telegram",
    -                "origin": {"platform": "telegram", "chat_id": "late"},
    -            }
    -        }))
    +        # The routing index now comes from _load_sessions_index() (state.db
    +        # primary, sessions.json fallback). Stub it to return the new
    +        # conversation, simulating the gateway having just written the
    +        # session row + first message in one state.db transaction.
    +        monkeypatch.setattr(
    +            mcp_serve, "_load_sessions_index",
    +            lambda: {
    +                "agent:main:telegram:dm:late": {
    +                    "session_id": session_id,
    +                    "platform": "telegram",
    +                    "origin": {"platform": "telegram", "chat_id": "late"},
    +                }
    +            },
    +        )
     
             class DB:
                 def get_messages(self, sid):
    @@ -1275,12 +1282,11 @@ def get_messages(self, sid):
                     }]
     
             bridge = mcp_serve.EventBridge()
    -        # Simulate the boundary state: state.db has NOT changed since the last
    -        # poll (its message landed on an earlier tick), but sessions.json was
    -        # only updated with this conversation now — the bridge has not yet seen
    -        # the current sessions.json content.
    -        bridge._state_db_mtime = db_path.stat().st_mtime
    -        bridge._sessions_json_mtime = 0.0
    +        # Bridge has never seen this db state (mtime differs) and has an
    +        # empty cached index — exactly the state after a new conversation's
    +        # first write.
    +        bridge._state_db_mtime = 0.0
    +        assert bridge._cached_sessions_index == {}
     
             bridge._poll_once(DB())