From 747386ecfa34eb5cbc10e104d326259fb538f565 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Sun, 5 Jul 2026 14:01:03 -0700 Subject: [PATCH] refactor: consolidate gateway session metadata into state.db (#58899) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- gateway/channel_directory.py | 62 +++++++++++- gateway/mirror.py | 28 +++++- gateway/run.py | 19 ++-- gateway/session.py | 57 +++++++++++ hermes_cli/status.py | 40 ++++++-- hermes_state.py | 189 ++++++++++++++++++++++++++++++++++- mcp_serve.py | 136 +++++++++++++++++++------ tests/test_hermes_state.py | 169 +++++++++++++++++++++++++++++++ tests/test_mcp_serve.py | 52 +++++----- 9 files changed, 671 insertions(+), 81 deletions(-) diff --git a/gateway/channel_directory.py b/gateway/channel_directory.py index 4a1bc151c26..16165d5e322 100644 --- a/gateway/channel_directory.py +++ b/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 [] diff --git a/gateway/mirror.py b/gateway/mirror.py index 164e371794d..44f528d6e04 100644 --- a/gateway/mirror.py +++ b/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 diff --git a/gateway/run.py b/gateway/run.py index 53562b4f483..656b03cce5a 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -7547,14 +7547,11 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew _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, @@ -7569,9 +7566,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew "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( diff --git a/gateway/session.py b/gateway/session.py index 5ced2d8aa78..48e32e31e39 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -1202,6 +1202,7 @@ class SessionStore: 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 @@ class SessionStore: 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 @@ class SessionStore: 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 @@ class SessionStore: 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 @@ class SessionStore: entry.session_id, session_key, entry.origin, + display_name=entry.display_name, ) def set_model_override( @@ -1888,6 +1943,7 @@ class SessionStore: 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 @@ class SessionStore: 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 diff --git a/hermes_cli/status.py b/hermes_cli/status.py index becc9403400..088460fb63f 100644 --- a/hermes_cli/status.py +++ b/hermes_cli/status.py @@ -543,17 +543,39 @@ def show_status(args): 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 diff --git a/hermes_state.py b/hermes_state.py index cbaa0205aa0..ab426f1e740 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -122,7 +122,7 @@ T = TypeVar("T") 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 @@ CREATE TABLE IF NOT EXISTS sessions ( 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 @@ class SessionDB: ) 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 @@ class SessionDB: 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 @@ class SessionDB: 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 @@ class SessionDB: 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, *, diff --git a/mcp_serve.py b/mcp_serve.py index 4ebda2253d7..558239153e6 100644 --- a/mcp_serve.py +++ b/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(): diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index eb884129bc5..e57a9286b6c 100644 --- a/tests/test_hermes_state.py +++ b/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") diff --git a/tests/test_mcp_serve.py b/tests/test_mcp_serve.py index 0ae403370d6..9c002925b20 100644 --- a/tests/test_mcp_serve.py +++ b/tests/test_mcp_serve.py @@ -1232,18 +1232,19 @@ class TestEventBridgePollE2E: 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. + """A brand-new conversation must be picked up on the tick where + state.db changes. - 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. + 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 @@ class TestEventBridgePollE2E: 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 @@ class TestEventBridgePollE2E: }] 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())