fix(gateway): respect reset boundaries during recovery (#68539)

find_latest_gateway_session_for_peer filtered non-recoverable rows out of
candidacy BEFORE ordering, so recovery could search behind a /new reset
boundary and resurrect an older still-open row for the same peer —
silently restoring the exact context the user reset.

Rebuilt against the #82633 finder (has-messages ranking +
COALESCE(last_activity_at, started_at) recency): the fence is expressed
as a NOT EXISTS guard inside both the exact-key and peer-fallback
queries — a candidate is rejected when an intentional boundary row
(session_reset / session_switch / idle / daily / suspended /
resume_pending_expired) for the same peer ended after the candidate's
last activity. If the conversation's most recent event is an intentional
reset, recovery returns nothing rather than reaching behind it.

Cherry-picked from #68617 and adapted to the rewritten finder.
(cherry picked from commit bb2c562a16)

hermes_state.py

@@ -4019,6 +4019,16 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         returning ``None`` mints a brand-new session id, which is a worse
         outcome than resuming an empty-but-correctly-keyed row (and "empty"
         may just mean the transcript lives under a compression child).
+
+        Reset boundaries fence recovery (#68539): an intentional boundary
+        such as ``session_reset`` (or any explicit non-recoverable
+        end_reason) must block fallback to an *older* row for the same
+        peer. Without the fence, the has-messages ranking above could reach
+        behind a /new reset and silently restore the exact context the user
+        reset. Each candidate is therefore rejected when a boundary row for
+        the peer ended *after* the candidate's last activity — if the
+        conversation's most recent event is an intentional reset, recovery
+        returns nothing rather than reaching behind it.
         """
         if not session_key:
             return None
@@ -4036,6 +4046,17 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                 WHERE s.session_key = ?
                   AND s.source = ?
                   AND (s.ended_at IS NULL OR s.end_reason IN ('agent_close', 'ws_orphan_reap'))
+                  AND NOT EXISTS (
+                      SELECT 1 FROM sessions b
+                      WHERE b.session_key = s.session_key
+                        AND b.source = s.source
+                        AND b.ended_at IS NOT NULL
+                        AND b.end_reason IN ('session_reset', 'session_switch',
+                                             'idle', 'daily', 'suspended',
+                                             'resume_pending_expired')
+                        AND b.ended_at
+                            > COALESCE(s.last_activity_at, s.started_at)
+                  )
                 ORDER BY _has_messages DESC,
                          COALESCE(s.last_activity_at, s.started_at) DESC
                 LIMIT 1
@@ -4069,6 +4090,20 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                   AND (COALESCE(s.message_count, 0) > 0 OR EXISTS (
                       SELECT 1 FROM messages WHERE messages.session_id = s.id LIMIT 1
                   ))
+                  AND NOT EXISTS (
+                      SELECT 1 FROM sessions b
+                      WHERE b.source = s.source
+                        AND COALESCE(b.user_id, '') = COALESCE(s.user_id, '')
+                        AND COALESCE(b.chat_id, '') = COALESCE(s.chat_id, '')
+                        AND COALESCE(b.chat_type, '') = COALESCE(s.chat_type, '')
+                        AND COALESCE(b.thread_id, '') = COALESCE(s.thread_id, '')
+                        AND b.ended_at IS NOT NULL
+                        AND b.end_reason IN ('session_reset', 'session_switch',
+                                             'idle', 'daily', 'suspended',
+                                             'resume_pending_expired')
+                        AND b.ended_at
+                            > COALESCE(s.last_activity_at, s.started_at)
+                  )
                 ORDER BY COALESCE(s.last_activity_at, s.started_at) DESC
                 LIMIT 1
                 """,

tests/gateway/test_session_store_stale_prune.py

@@ -126,6 +126,48 @@ class TestPruneStaleSessionsLocked:
             store._prune_stale_sessions_locked()
             mock_save.assert_called_once()
 
+    def test_reset_boundary_does_not_recover_older_session_for_peer(self, tmp_path):
+        """Startup pruning must not search past an intentional reset boundary.
+
+        The durable recovery query deliberately excludes ``session_reset``
+        rows — and a newer reset row must also fence any *older* still-open
+        row for the same peer. If startup pruning invokes recovery for a
+        routing entry that points at such a row, the query must not return
+        an older live session for the same peer and silently restore the
+        context that the user reset. Exercise the real SessionDB query here
+        rather than mocking its result.
+        """
+        from hermes_state import SessionDB
+
+        key = "agent:main:telegram:dm:5140768830"
+        db = SessionDB(tmp_path / "state.db")
+        peer = {
+            "user_id": "5140768830",
+            "session_key": key,
+            "chat_id": "5140768830",
+            "chat_type": "dm",
+        }
+        db.create_session("sid_before_reset", "telegram", **peer)
+        db.append_message("sid_before_reset", "user", "private old context")
+        db.create_session("sid_reset", "telegram", **peer)
+        db.append_message("sid_reset", "user", "/new")
+        db.end_session("sid_reset", "session_reset")
+
+        store = _make_store_with_db(tmp_path / "sessions", db)
+        stale_entry = _make_entry_with_origin(key, "sid_reset")
+        store._entries[key] = stale_entry
+
+        # Model restart startup followed by the peer's first incoming message.
+        store._prune_stale_sessions_locked()
+        assert stale_entry.origin is not None
+        current = store.get_or_create_session(stale_entry.origin)
+
+        assert current.session_id not in {"sid_before_reset", "sid_reset"}
+        assert store._entries[key].session_id == current.session_id
+        reset_row = db.get_session("sid_reset")
+        assert reset_row is not None
+        assert reset_row["end_reason"] == "session_reset"
+
 
 # ---------------------------------------------------------------------------
 # Integration: _ensure_loaded_locked calls _prune_stale_sessions_locked

tests/test_hermes_state.py

@@ -3496,6 +3496,41 @@ def test_gateway_session_peer_round_trip_and_recovery(db):
     assert recovered["id"] == "gw-session"
 
 
+@pytest.mark.parametrize(
+    "persisted_session_key",
+    ["agent:main:telegram:dm:chat-1", None],
+    ids=["exact-key", "peer-fallback"],
+)
+def test_gateway_session_recovery_does_not_cross_newer_reset_boundary(
+    db, persisted_session_key
+):
+    """A newer session_reset row fences recovery for the peer (#68539).
+
+    Recovery must never reach *behind* an intentional /new boundary and
+    resurrect an older still-open row — if the newest boundary row for the
+    peer is reset-ended, recovery returns nothing.
+    """
+    peer = {
+        "user_id": "user-1",
+        "session_key": persisted_session_key,
+        "chat_id": "chat-1",
+        "chat_type": "dm",
+    }
+    db.create_session("gw-before-reset", "telegram", **peer)
+    db.append_message("gw-before-reset", "user", "old context")
+    db.create_session("gw-reset", "telegram", **peer)
+    db.append_message("gw-reset", "user", "/new")
+    db.end_session("gw-reset", "session_reset")
+
+    assert db.find_latest_gateway_session_for_peer(
+        source="telegram",
+        user_id="user-1",
+        session_key="agent:main:telegram:dm:chat-1",
+        chat_id="chat-1",
+        chat_type="dm",
+    ) is None
+
+