fix(gateway): prune stale sessions.json entries on startup (FM9)

gateway/session.py

@@ -847,7 +847,60 @@ def _ensure_loaded_locked(self) -> None:
                 print(f"[gateway] Warning: Failed to load sessions: {e}")
 
         self._loaded = True
-
+
+        # Prune any sessions.json entries that point to ended sessions in the
+        # DB.  This self-heals the FM9 failure mode: a gateway crash (exit code
+        # 1) skips the normal shutdown path so sessions.json is never cleared.
+        # On the next startup those stale entries act as live routing keys, but
+        # every incoming message is silently dropped because the session is
+        # already closed.  Running this here (with _lock already held) is safe
+        # and cheap — one SELECT per routing key, done exactly once at startup.
+        self._prune_stale_sessions_locked()
+
+    def _prune_stale_sessions_locked(self) -> None:
+        """Remove sessions.json entries whose session has ended in state.db.
+
+        Called once during startup (from _ensure_loaded_locked, lock held).
+        Guards against the FM9 failure mode where a crashed gateway leaves
+        stale routing entries that silently drop all incoming messages.
+
+        A ``session_id`` is stale when state.db shows ``end_reason IS NOT NULL``
+        for it.  Sessions absent from the DB (never persisted) are left alone so
+        that pre-DB legacy gateways are unaffected.  ``self._db`` being None
+        (SQLite unavailable) is also a no-op.
+        """
+        db = getattr(self, "_db", None)
+        if not db or not self._entries:
+            return
+
+        stale_keys: list[str] = []
+        try:
+            for key, entry in self._entries.items():
+                session_row = db.get_session(entry.session_id)
+                # session_row is None  → not in DB (legacy / pre-SQLite) — keep
+                # end_reason is None  → session alive — keep
+                # end_reason is not None → session ended — prune
+                if session_row is not None and session_row.get("end_reason") is not None:
+                    logger.warning(
+                        "gateway.session: pruning stale sessions.json entry"
+                        " %r → %s (end_reason=%r); left by a crashed gateway",
+                        key,
+                        entry.session_id,
+                        session_row["end_reason"],
+                    )
+                    stale_keys.append(key)
+        except Exception as exc:  # pragma: no cover — DB errors are non-fatal here
+            logger.warning(
+                "gateway.session: stale-entry pruning skipped due to DB error: %s", exc
+            )
+            return
+
+        for key in stale_keys:
+            del self._entries[key]
+
+        if stale_keys:
+            self._save()
+
     def _save(self) -> None:
         """Save sessions index to disk (kept for session key -> ID mapping)."""
         import tempfile

scripts/release.py

@@ -1696,6 +1696,7 @@
     "qs2816661685@gmail.com": "qingshan89",  # PR #46895 co-author (desktop remote artifact download)
     "yspdev@gmail.com": "AJ",  # PR #44510 co-author (desktop named-profile boot loop)
     "steveonjava@gmail.com": "steveonjava",  # PR #29669 (redact secrets in kanban tool payloads)
+    "hephaestus@hermes.ai": "terry197913",  # PR #52808 (FM9 sessions.json prune)
 }
 
 

tests/gateway/test_session_store_stale_prune.py

@@ -0,0 +1,231 @@
+"""Tests for SessionStore._prune_stale_sessions_locked — FM9 self-healing.
+
+When a gateway crashes (exit code 1) the normal shutdown path is skipped and
+sessions.json is left pointing at sessions that are already ended in state.db.
+On the next startup _ensure_loaded_locked calls _prune_stale_sessions_locked to
+detect and remove those stale routing entries before they can silently drop
+incoming messages.
+
+Failure mode being tested (FM9):
+  - Gateway crashes mid-run → sessions.json not cleared
+  - state.db marks the session ended (end_reason IS NOT NULL)
+  - Gateway restarts → trusts sessions.json → every message silently dropped
+  - Fix: prune stale entries on startup
+"""
+
+import json
+import threading
+from datetime import datetime, timedelta
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from gateway.config import GatewayConfig, Platform, SessionResetPolicy
+from gateway.session import SessionEntry, SessionStore
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _make_entry(key: str, session_id: str) -> SessionEntry:
+    now = datetime.now()
+    return SessionEntry(
+        session_key=key,
+        session_id=session_id,
+        created_at=now - timedelta(hours=2),
+        updated_at=now - timedelta(hours=1),
+        platform=Platform.TELEGRAM,
+        chat_type="dm",
+    )
+
+
+def _make_store_with_db(tmp_path, db_mock) -> SessionStore:
+    """Build a SessionStore with a mock SessionDB, bypassing disk load."""
+    config = GatewayConfig(
+        default_reset_policy=SessionResetPolicy(mode="none"),
+    )
+    with patch("gateway.session.SessionStore._ensure_loaded"):
+        store = SessionStore(sessions_dir=tmp_path, config=config)
+    store._db = db_mock
+    store._loaded = True
+    return store
+
+
+def _db_returning(rows: dict) -> MagicMock:
+    """Return a SessionDB mock where get_session maps session_id → row dict."""
+    db = MagicMock()
+    db.get_session.side_effect = lambda sid: rows.get(sid)
+    return db
+
+
+# ---------------------------------------------------------------------------
+# Core behaviour
+# ---------------------------------------------------------------------------
+
+class TestPruneStaleSessionsLocked:
+    def test_prunes_ended_session(self, tmp_path):
+        """Entry pointing at a session with end_reason is removed."""
+        db = _db_returning({
+            "sid_dm": {"end_reason": "agent_close", "id": "sid_dm"},
+        })
+        store = _make_store_with_db(tmp_path, db)
+        store._entries["dm_key"] = _make_entry("dm_key", "sid_dm")
+
+        store._prune_stale_sessions_locked()
+
+        assert "dm_key" not in store._entries
+
+    def test_keeps_live_session(self, tmp_path):
+        """Entry pointing at a session with end_reason=None is kept."""
+        db = _db_returning({
+            "sid_live": {"end_reason": None, "id": "sid_live"},
+        })
+        store = _make_store_with_db(tmp_path, db)
+        store._entries["live_key"] = _make_entry("live_key", "sid_live")
+
+        store._prune_stale_sessions_locked()
+
+        assert "live_key" in store._entries
+
+    def test_keeps_session_absent_from_db(self, tmp_path):
+        """Entry for a session_id not in state.db (legacy) is left alone."""
+        db = _db_returning({})  # empty — session never in DB
+        store = _make_store_with_db(tmp_path, db)
+        store._entries["legacy_key"] = _make_entry("legacy_key", "sid_legacy")
+
+        store._prune_stale_sessions_locked()
+
+        assert "legacy_key" in store._entries
+
+    def test_prunes_multiple_stale_entries(self, tmp_path):
+        """All stale entries are removed in one pass."""
+        db = _db_returning({
+            "sid_a": {"end_reason": "agent_close", "id": "sid_a"},
+            "sid_b": {"end_reason": "session_reset", "id": "sid_b"},
+            "sid_c": {"end_reason": None, "id": "sid_c"},  # alive — keep
+        })
+        store = _make_store_with_db(tmp_path, db)
+        store._entries["key_a"] = _make_entry("key_a", "sid_a")
+        store._entries["key_b"] = _make_entry("key_b", "sid_b")
+        store._entries["key_c"] = _make_entry("key_c", "sid_c")
+
+        store._prune_stale_sessions_locked()
+
+        assert "key_a" not in store._entries
+        assert "key_b" not in store._entries
+        assert "key_c" in store._entries
+
+    def test_noop_when_db_is_none(self, tmp_path):
+        """If SQLite is unavailable (_db=None) pruning is silently skipped."""
+        config = GatewayConfig(
+            default_reset_policy=SessionResetPolicy(mode="none"),
+        )
+        with patch("gateway.session.SessionStore._ensure_loaded"):
+            store = SessionStore(sessions_dir=tmp_path, config=config)
+        store._db = None
+        store._loaded = True
+        store._entries["key"] = _make_entry("key", "sid_x")
+
+        store._prune_stale_sessions_locked()  # must not raise
+
+        assert "key" in store._entries  # unchanged
+
+    def test_noop_when_no_entries(self, tmp_path):
+        """Empty _entries dict → no DB calls, no error."""
+        db = MagicMock()
+        store = _make_store_with_db(tmp_path, db)
+
+        store._prune_stale_sessions_locked()
+
+        db.get_session.assert_not_called()
+
+    def test_db_error_is_non_fatal(self, tmp_path):
+        """A DB exception during pruning must not crash the gateway startup."""
+        db = MagicMock()
+        db.get_session.side_effect = Exception("DB locked")
+        store = _make_store_with_db(tmp_path, db)
+        store._entries["key"] = _make_entry("key", "sid_x")
+
+        store._prune_stale_sessions_locked()  # must not raise
+
+        # Entry is left intact (safe fallback)
+        assert "key" in store._entries
+
+    def test_sessions_json_rewritten_after_pruning(self, tmp_path):
+        """sessions.json must be updated after stale entries are removed."""
+        db = _db_returning({
+            "sid_stale": {"end_reason": "agent_close", "id": "sid_stale"},
+        })
+        store = _make_store_with_db(tmp_path, db)
+        store._entries["stale_key"] = _make_entry("stale_key", "sid_stale")
+
+        with patch.object(store, "_save") as mock_save:
+            store._prune_stale_sessions_locked()
+            mock_save.assert_called_once()
+
+    def test_sessions_json_not_rewritten_when_nothing_pruned(self, tmp_path):
+        """_save() must NOT be called when no entries are removed."""
+        db = _db_returning({
+            "sid_live": {"end_reason": None, "id": "sid_live"},
+        })
+        store = _make_store_with_db(tmp_path, db)
+        store._entries["live_key"] = _make_entry("live_key", "sid_live")
+
+        with patch.object(store, "_save") as mock_save:
+            store._prune_stale_sessions_locked()
+            mock_save.assert_not_called()
+
+
+# ---------------------------------------------------------------------------
+# Integration: _ensure_loaded_locked calls _prune_stale_sessions_locked
+# ---------------------------------------------------------------------------
+
+class TestEnsureLoadedCallsPrune:
+    def test_prune_called_during_load(self, tmp_path):
+        """_prune_stale_sessions_locked must be invoked at the end of startup."""
+        # Write a sessions.json with one stale entry
+        entry = _make_entry("dm_key", "sid_stale")
+        sessions_file = tmp_path / "sessions.json"
+        sessions_file.write_text(
+            json.dumps({"dm_key": entry.to_dict()}, indent=2),
+            encoding="utf-8",
+        )
+
+        db = _db_returning({
+            "sid_stale": {"end_reason": "agent_close", "id": "sid_stale"},
+        })
+        config = GatewayConfig(
+            default_reset_policy=SessionResetPolicy(mode="none"),
+        )
+        store = SessionStore(sessions_dir=tmp_path, config=config)
+        store._db = db
+
+        # Trigger the real load path (not mocked)
+        store._ensure_loaded()
+
+        assert "dm_key" not in store._entries, (
+            "Stale entry must be pruned during _ensure_loaded"
+        )
+
+    def test_live_entry_survives_load(self, tmp_path):
+        """A live sessions.json entry must not be removed during load."""
+        entry = _make_entry("active_key", "sid_live")
+        sessions_file = tmp_path / "sessions.json"
+        sessions_file.write_text(
+            json.dumps({"active_key": entry.to_dict()}, indent=2),
+            encoding="utf-8",
+        )
+
+        db = _db_returning({
+            "sid_live": {"end_reason": None, "id": "sid_live"},
+        })
+        config = GatewayConfig(
+            default_reset_policy=SessionResetPolicy(mode="none"),
+        )
+        store = SessionStore(sessions_dir=tmp_path, config=config)
+        store._db = db
+
+        store._ensure_loaded()
+
+        assert "active_key" in store._entries