fix(gateway): notify resets finalized through the recovery fence

gateway/run.py

@@ -18770,6 +18770,12 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
         # If the previous session expired and was auto-reset, deliver a notice
         # so the agent knows this is a fresh conversation (not an intentional /reset).
         if _was_auto_reset:
+            # The `or 'idle'` default is a two-way fallback: it covers a
+            # legacy entry whose auto_reset_reason predates the field, AND
+            # any future policy mode this mapping has no branch for —
+            # unknown reasons intentionally degrade to the generic idle
+            # notice rather than skipping the fresh-session note (review
+            # on #89314). Add an explicit branch when a new mode lands.
             reset_reason = getattr(session_entry, 'auto_reset_reason', None) or 'idle'
             if reset_reason == "suspended":
                 context_note = "[System note: The user's previous session was stopped and suspended. This is a fresh conversation with no prior context.]"

gateway/session.py

@@ -2825,6 +2825,44 @@ def _get_or_create_session_impl(
                     entry = published
                     _needs_save = True
 
+        # Recovery can also come back empty because the #68539 reset fence
+        # blocked it: the peer's latest durable row is an intentional
+        # ``session_reset`` boundary (the expiry watcher finalized it while the
+        # sessions.json entry went missing or stale). Nothing above sets
+        # was_auto_reset in that shape, so the user-facing "session
+        # automatically reset" notice is silently dropped even though the
+        # reset genuinely happened (#89314). Recover the boundary and flag the
+        # reset like the live-entry paths do. A manual /reset is not confused
+        # with this: it mints a fresh live row that recovery would have found,
+        # so reaching here with a session_reset boundary means the reset was
+        # never followed by a new conversation yet.
+        if entry is None and not was_auto_reset:
+            boundary_finder = getattr(
+                self._db, "find_latest_reset_boundary_for_peer", None
+            )
+            if callable(boundary_finder):
+                try:
+                    boundary = boundary_finder(
+                        session_key=session_key,
+                        source=source.platform.value,
+                    )
+                except Exception as exc:
+                    logger.debug(
+                        "Gateway session reset-boundary lookup failed for %s: %s",
+                        session_key, exc,
+                    )
+                    boundary = None
+                if boundary and boundary.get("end_reason") == "session_reset":
+                    policy = self.config.get_reset_policy(
+                        platform=source.platform,
+                        session_type=source.chat_type,
+                    )
+                    if policy.mode != "none":
+                        was_auto_reset = True
+                        auto_reset_reason = "daily" if policy.mode == "daily" else "idle"
+                        reset_had_activity = bool(boundary.get("message_count"))
+                        prev_session_id = boundary.get("id")
+
         if entry is None:
             # Create a candidate outside the lock, then publish only if another
             # worker has not already populated this routing key.

hermes_state.py

@@ -5265,6 +5265,38 @@ def find_latest_gateway_session_for_peer(
             ).fetchone()
         return self._session_row_dict(row) if row else None
 
+    def find_latest_reset_boundary_for_peer(
+        self,
+        *,
+        session_key: str,
+        source: str,
+    ) -> Optional[Dict[str, Any]]:
+        """Return the peer's most recent intentional reset boundary row.
+
+        Recovery fences on these rows (#68539): when the fence is the reason
+        ``find_latest_gateway_session_for_peer`` returned nothing, the caller
+        still needs the boundary itself so it can tell the user their session
+        was auto-reset instead of silently starting fresh (#89314).
+        """
+        if not session_key:
+            return None
+        with self._lock:
+            row = self._conn.execute(
+                f"""
+                SELECT id, end_reason, ended_at,
+                       COALESCE(last_activity_at, started_at) AS last_activity,
+                       COALESCE(message_count, 0) AS message_count
+                FROM sessions
+                WHERE session_key = ?
+                  AND source = ?
+                  AND end_reason IN ({_RESET_END_REASONS_SQL})
+                ORDER BY ended_at DESC
+                LIMIT 1
+                """,
+                (session_key, source),
+            ).fetchone()
+        return dict(row) if row is not None else None
+
     # ── Orphaned gateway-session repair (#82616) ──────────────────────────
     # A write-path failure (corrupt FTS, crash between routing publication
     # and row creation) can leave the live conversation in a session row

tests/gateway/test_session_store_runtime_stale_guard.py

@@ -314,3 +314,88 @@ def test_repoint_does_not_touch_activity_clock(self, tmp_path):
         assert store.suspend_recently_active(max_age_seconds=120) == 0
 
 
+class TestResetBoundaryNotice:
+    """A reset finalized through the #68539 fence must still notify (#89314).
+
+    The expiry watcher ends the session in state.db; if the sessions.json
+    entry is missing or stale by the time the next message arrives, recovery
+    returns None (the boundary fences it) and the fresh-session path used to
+    create a new session with was_auto_reset=False — silently dropping the
+    "session automatically reset" notice the live-entry path delivers."""
+
+    def test_fenced_reset_boundary_flags_auto_reset_on_fresh_session(
+        self, tmp_path,
+    ):
+        source = _source()
+        config = GatewayConfig(
+            default_reset_policy=SessionResetPolicy(mode="idle", idle_minutes=60),
+        )
+        db = _db_returning({})
+        # Recovery finds nothing (the boundary fences it)...
+        db.find_latest_gateway_session_for_peer.return_value = None
+        # ...but the peer's latest durable row is a session_reset boundary.
+        db.find_latest_reset_boundary_for_peer.return_value = {
+            "id": "sid_reset",
+            "end_reason": "session_reset",
+            "ended_at": datetime.now().timestamp() - 60,
+            "last_activity": datetime.now().timestamp() - 3600,
+            "message_count": 41,
+        }
+        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 routing index lost the mapping.
+
+        result = store.get_or_create_session(source)
+
+        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_reset"
+
+    def test_explicit_new_command_boundary_never_flags_auto_reset(
+        self, tmp_path,
+    ):
+        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 = None
+        # /new and other explicit boundaries are not automatic resets.
+        db.find_latest_reset_boundary_for_peer.return_value = {
+            "id": "sid_new",
+            "end_reason": "new_command",
+            "ended_at": datetime.now().timestamp() - 60,
+            "last_activity": datetime.now().timestamp() - 3600,
+            "message_count": 5,
+        }
+        with patch("gateway.session.SessionStore._ensure_loaded"):
+            store = SessionStore(sessions_dir=tmp_path, config=config)
+        store._db = db
+        store._loaded = True
+
+        result = store.get_or_create_session(source)
+
+        assert result.was_auto_reset is False
+        assert result.auto_reset_reason is None
+
+    def test_mode_none_never_flags_auto_reset(self, tmp_path):
+        source = _source()
+        db = _db_returning({})
+        db.find_latest_gateway_session_for_peer.return_value = None
+        db.find_latest_reset_boundary_for_peer.return_value = {
+            "id": "sid_reset",
+            "end_reason": "session_reset",
+            "ended_at": datetime.now().timestamp() - 60,
+            "last_activity": datetime.now().timestamp() - 3600,
+            "message_count": 3,
+        }
+        store = _make_store_with_db(tmp_path, db)  # mode="none"
+
+        result = store.get_or_create_session(source)
+
+        assert result.was_auto_reset is False
+
+