fix(gateway): session recovery honors reset boundaries and real idle time (#68617 + #78618 salvage)

contributors/emails/hill.chitsanupong@gmail.com

@@ -0,0 +1 @@
+hillimited

gateway/session.py

@@ -1903,18 +1903,37 @@ def _create_entry_from_recovered_row(
     ) -> SessionEntry:
         started_at = row.get("started_at")
         try:
-            created_at = datetime.fromtimestamp(float(started_at)) if started_at else now
+            created_at = datetime.fromtimestamp(float(started_at))
         except (TypeError, ValueError, OSError):
-            created_at = now
+            # An invalid durable timestamp must look old, never freshly active.
+            created_at = datetime.fromtimestamp(0)
+        # The finder already returns the row's durable recency
+        # (last_activity_at is what it ranks candidates by), so no extra DB
+        # round-trip is needed: derive updated_at straight from the row.
+        last_activity = row.get("last_activity_at")
+        try:
+            updated_at = (
+                datetime.fromtimestamp(float(last_activity))
+                if last_activity is not None
+                else created_at
+            )
+        except (TypeError, ValueError, OSError):
+            updated_at = created_at
+        had_activity = row.get("_has_messages")
+        if had_activity is None:
+            had_activity = bool(row.get("message_count") or 0) or (
+                last_activity is not None
+            )
         return SessionEntry(
             session_key=session_key,
             session_id=str(row["id"]),
             created_at=created_at,
-            updated_at=now,
+            updated_at=updated_at,
             origin=source,
             display_name=source.chat_name,
             platform=source.platform,
             chat_type=source.chat_type,
+            reset_had_activity=bool(had_activity),
         )
 
     def _find_gateway_session_row(
@@ -1964,7 +1983,13 @@ def _recover_session_from_db(
         now: datetime,
         raise_on_lookup_error: bool = False,
     ) -> Optional[SessionEntry]:
-        """Rebuild a missing session-key mapping from durable state.db data."""
+        """Rebuild a missing session-key mapping from durable state.db data.
+
+        Returns ``None`` when no row is recoverable, or when the recovered
+        session is already overdue under the configured reset policy — the
+        row is then durably promoted to a reset boundary instead of being
+        resurrected as freshly active.
+        """
         legacy_key = self._legacy_slack_session_key(source)
         recovered = self._find_gateway_session_row(
             session_key=session_key,
@@ -2001,16 +2026,31 @@ def _recover_session_from_db(
                 session_key,
             )
             return None
-        try:
-            self._db.reopen_session(str(recovered["id"]))
-        except Exception as exc:
-            logger.debug("Gateway session DB reopen failed for %s: %s", session_key, exc)
         entry = self._create_entry_from_recovered_row(
             row=recovered,
             session_key=session_key,
             source=source,
             now=now,
         )
+        reset_reason = self._should_reset(entry, source)
+        if reset_reason:
+            try:
+                promote = getattr(self._db, "promote_to_session_reset", None)
+                if callable(promote):
+                    promote(entry.session_id, reset_reason)
+                else:
+                    self._db.end_session(entry.session_id, reset_reason)
+            except Exception as exc:
+                logger.debug(
+                    "Gateway recovered-session reset promotion failed for %s: %s",
+                    session_key,
+                    exc,
+                )
+            return None
+        try:
+            self._db.reopen_session(entry.session_id)
+        except Exception as exc:
+            logger.debug("Gateway session DB reopen failed for %s: %s", session_key, exc)
         if migrated_legacy:
             self._record_gateway_session_peer(
                 entry.session_id,
@@ -2026,6 +2066,8 @@ def _query_recoverable_session(
         """DB-only half of _recover_session_from_db (no lock needed).
 
         Returns a SessionEntry or None.  Caller assigns _entries[key] under lock.
+        The returned entry's session row is NOT reopened here: the caller
+        evaluates the reset policy first and decides reset vs resume.
         """
         legacy_key = self._legacy_slack_session_key(source)
         recovered = self._find_gateway_session_row(
@@ -2061,11 +2103,9 @@ def _query_recoverable_session(
                 session_key,
             )
             return None
-        try:
-            self._db.reopen_session(str(recovered["id"]))
-        except Exception as exc:
-            logger.debug("Gateway session DB reopen failed for %s: %s",
-                         session_key, exc)
+        # Reopen only after the caller evaluates reset policy against durable
+        # last activity.  An agent_close/ws_orphan row may need promotion to a
+        # real reset boundary instead.
         entry = self._create_entry_from_recovered_row(
             row=recovered, session_key=session_key, source=source, now=now,
         )
@@ -2638,13 +2678,29 @@ def _get_or_create_session_impl(
                 session_key=session_key, source=source, now=now,
             )
             if recovered is not None:
-                with self._lock:
-                    published = self._entries.get(session_key)
-                    if published is None:
-                        self._entries[session_key] = recovered
-                        published = recovered
-                entry = published
-                _needs_save = True
+                recovered_reset_reason = self._should_reset(recovered, source)
+                if recovered_reset_reason:
+                    was_auto_reset = True
+                    auto_reset_reason = recovered_reset_reason
+                    reset_had_activity = recovered.reset_had_activity
+                    db_end_session_id = recovered.session_id
+                    prev_session_id = recovered.session_id
+                else:
+                    try:
+                        self._db.reopen_session(recovered.session_id)
+                    except Exception as exc:
+                        logger.debug(
+                            "Gateway session DB reopen failed for %s: %s",
+                            session_key,
+                            exc,
+                        )
+                    with self._lock:
+                        published = self._entries.get(session_key)
+                        if published is None:
+                            self._entries[session_key] = recovered
+                            published = recovered
+                    entry = published
+                    _needs_save = True
 
         if entry is None:
             # Create a candidate outside the lock, then publish only if another

hermes_state.py

@@ -4019,6 +4019,16 @@ def find_latest_gateway_session_for_peer(
         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 @@ def find_latest_gateway_session_for_peer(
                 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 @@ def find_latest_gateway_session_for_peer(
                   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_runtime_stale_guard.py

@@ -165,6 +165,111 @@ def test_stale_agent_close_overdue_policy_creates_fresh_session(
         db.create_session.assert_called_once()
 
 
+class TestRecoveredSessionResetPolicy:
+    """Recovery must not resurrect sessions as freshly active.
+
+    ``_create_entry_from_recovered_row`` used to stamp ``updated_at=now`` on
+    the rebuilt entry, so an opt-in idle/daily ``session_reset`` policy could
+    never fire across a gateway restart: the recovered session always looked
+    freshly active, and every subsequent message bumped ``updated_at`` again
+    — a recovered stale session could never age out. The entry now carries
+    the durable ``last_activity_at`` the finder already returns on the row
+    and the recovery paths evaluate ``_should_reset`` before resuming.
+    """
+
+    def test_recovered_entry_carries_durable_last_activity(self, tmp_path):
+        """A recovered mapping reports the DB's last message time, not now()."""
+        source = _source()
+        started = (datetime.now() - timedelta(hours=3)).timestamp()
+        last_activity = (datetime.now() - timedelta(hours=2)).timestamp()
+        db = _db_returning({})
+        db.find_latest_gateway_session_for_peer.return_value = {
+            "id": "sid_recovered",
+            "started_at": started,
+            "last_activity_at": last_activity,
+        }
+        store = _make_store_with_db(tmp_path, db)  # default mode="none"
+
+        result = store.get_or_create_session(source)
+
+        assert result.session_id == "sid_recovered"
+        assert result.created_at == datetime.fromtimestamp(started)
+        assert result.updated_at == datetime.fromtimestamp(last_activity)
+        assert result.reset_had_activity is True
+
+    def test_recovered_session_past_idle_policy_resets_instead_of_resuming(
+        self, tmp_path,
+    ):
+        """Lost mapping + overdue recoverable row → reset, not silent resume."""
+        source = _source()
+        config = GatewayConfig(
+            default_reset_policy=SessionResetPolicy(mode="idle", idle_minutes=60),
+        )
+        db = _db_returning({})
+        db.find_latest_gateway_session_for_peer.return_value = {
+            "id": "sid_idle",
+            "started_at": (datetime.now() - timedelta(hours=3)).timestamp(),
+            "last_activity_at": (
+                datetime.now() - timedelta(hours=2)
+            ).timestamp(),
+        }
+        with patch("gateway.session.SessionStore._ensure_loaded"):
+            store = SessionStore(sessions_dir=tmp_path, config=config)
+        store._db = db
+        store._loaded = True
+        # No in-memory entry: the mapping was lost (e.g. crash before save).
+
+        result = store.get_or_create_session(source)
+
+        assert result.session_id != "sid_idle"
+        assert result.was_auto_reset is True
+        assert result.auto_reset_reason == "idle"
+        assert result.reset_had_activity is True
+        assert result.prev_session_id == "sid_idle"
+        db.reopen_session.assert_not_called()
+        db.promote_to_session_reset.assert_called_once_with("sid_idle", "idle")
+        db.end_session.assert_not_called()
+        db.create_session.assert_called_once()
+
+    def test_default_none_policy_recovery_resumes_unchanged(self, tmp_path):
+        """mode="none" (the default) still resumes recoverable rows as before."""
+        source = _source()
+        db = _db_returning({})
+        db.find_latest_gateway_session_for_peer.return_value = {
+            "id": "sid_recovered",
+            "started_at": (datetime.now() - timedelta(days=30)).timestamp(),
+            "last_activity_at": (
+                datetime.now() - timedelta(days=30)
+            ).timestamp(),
+        }
+        store = _make_store_with_db(tmp_path, db)  # default mode="none"
+
+        result = store.get_or_create_session(source)
+
+        assert result.session_id == "sid_recovered"
+        db.reopen_session.assert_called_once_with("sid_recovered")
+        db.promote_to_session_reset.assert_not_called()
+        db.end_session.assert_not_called()
+        db.create_session.assert_not_called()
+
+    def test_recovery_tolerates_row_without_last_activity(self, tmp_path):
+        """A row lacking last_activity_at falls back to created_at."""
+        source = _source()
+        started = (datetime.now() - timedelta(hours=3)).timestamp()
+        db = _db_returning({})
+        db.find_latest_gateway_session_for_peer.return_value = {
+            "id": "sid_recovered",
+            "started_at": started,
+        }
+        store = _make_store_with_db(tmp_path, db)
+
+        result = store.get_or_create_session(source)
+
+        assert result.session_id == "sid_recovered"
+        assert result.updated_at == datetime.fromtimestamp(started)
+        assert result.reset_had_activity is False
+
+
 class TestAdvanceCompressionSession:
     def test_cas_advances_route_without_reopening_rows(self, tmp_path):
         db = _db_returning({})