fix(state): filter the started_at fallback of last_active to the epoch window

A session whose started_at is corrupt and has no in-window activity or
message timestamp still returned the raw cell as last_active, and the
order_by_last_active fallback sorted it above every healthy session.
Both fallbacks now go through the same window as the UNION ALL values;
a session with no trusted timestamp gets NULL.

hermes_state_common.py

@@ -217,16 +217,24 @@ def _ephemeral_child_sql(alias: str = "s") -> str:
         f" AND NOT ({_COMPRESSION_CHILD_SQL.format(a=alias)}) AND NOT ({_RESET_CHILD_SQL.format(a=alias)}))")
 
 
+_SQL_IN_WINDOW = f"BETWEEN {EPOCH_MIN!r} AND {EPOCH_MAX!r}"
+
+
+def _sql_in_window(expr: str) -> str:
+    """*expr* when it is inside the ``coerce_epoch`` window, else NULL."""
+    return f"(SELECT _win.v FROM (SELECT {expr} AS v) _win WHERE _win.v {_SQL_IN_WINDOW})"
+
+
 def _sql_freshest_of(activity: str, session_id_expr: str, started: str) -> str:
     """Freshest of *activity* and the latest message timestamp for *session_id_expr*, else *started*.
     Heartbeats are rate-limited (~60s) so ``last_activity_at`` can lag a newer message; never use it alone.
     Cells outside the ``coerce_epoch`` window (garbage doubles salvaged from a damaged page, TEXT) are
-    skipped, or one bad message row pins the session's recency to ``5e+246`` (#91536)."""
-    in_window = f"BETWEEN {EPOCH_MIN!r} AND {EPOCH_MAX!r}"
+    skipped, fallback included, or one bad row pins the session's recency to ``5e+246`` (#91536); a
+    session with no trusted cell at all is NULL."""
     msg_max = (f"(SELECT MAX(_act_m.timestamp) FROM messages _act_m WHERE _act_m.session_id = {session_id_expr}"
-        f" AND _act_m.timestamp {in_window})")
+        f" AND _act_m.timestamp {_SQL_IN_WINDOW})")
     return (f"COALESCE((SELECT MAX(_act_v.v) FROM (SELECT {activity} AS v UNION ALL SELECT {msg_max}) _act_v"
-        f" WHERE _act_v.v {in_window}), {started})")
+        f" WHERE _act_v.v {_SQL_IN_WINDOW}), {_sql_in_window(started)})")
 
 
 def _sql_session_last_active(alias: str = "s") -> str:

hermes_state_sessions.py

@@ -18,8 +18,8 @@ from hermes_startup_watchdog import report_startup_progress
 from hermes_state_common import (
     _LISTABLE_CHILD_SQL, _PREVIEW_ELIGIBLE_SQL, _PREVIEW_RAW_SELECT, _RECOVERABLE_END_REASONS,
     _RECOVERABLE_END_REASONS_SQL, _RESET_CHILD_SQL, _RESET_END_REASONS, _legacy_reset_child_sql, _shape_preview,
-    _sql_json_extract, _sql_session_last_active, _sql_session_last_active_by_id, escape_like as _escape_like,
-    _SQL_IN_CHUNK, _id_chunks, _placeholders as _session_ids_placeholders,
+    _sql_in_window, _sql_json_extract, _sql_session_last_active, _sql_session_last_active_by_id,
+    escape_like as _escape_like, _SQL_IN_CHUNK, _id_chunks, _placeholders as _session_ids_placeholders,
 )
 
 # caplog tests pin the "hermes_state" logger name.
@@ -1323,7 +1323,7 @@ class SessionSessionsMixin:
                     GROUP BY root_id
                 )
                 {select_head}{_sql_session_last_active("s")} AS last_active,
-                    COALESCE(cm.effective_last_active, s.started_at) AS _effective_last_active
+                    COALESCE(cm.effective_last_active, {_sql_in_window("s.started_at")}) AS _effective_last_active
                 FROM sessions s
                 LEFT JOIN chain_max cm ON cm.root_id = s.id
                 {prompt_join}

tests/hermes_state/test_corrupt_row_robustness.py

@@ -80,6 +80,21 @@ def test_last_active_skips_a_garbage_message_timestamp(tmp_path):
     assert tip_rows["recovered"]["last_active"] == good
 
 
+def test_last_active_never_falls_back_to_a_corrupt_started_at(corrupt_db):
+    """With no in-window activity or message timestamp left, the ``started_at`` fallback must be filtered
+    too, or the corrupt cell becomes ``last_active`` and pins the session to the top of MRU (#91536)."""
+    corrupt_db._execute_write(lambda conn: conn.execute(
+        "UPDATE sessions SET last_activity_at = NULL WHERE id = 'bad-huge'"))
+    listings = (corrupt_db.list_sessions_rich(limit=10), corrupt_db.list_sessions_rich(limit=10, order_by_last_active=True),
+                corrupt_db.search_sessions(limit=10))
+    for rows in listings:
+        by_id = {r["id"]: r for r in rows}
+        assert by_id["bad-huge"]["last_active"] is None and by_id["bad-text"]["last_active"] is None
+        assert by_id["good"]["last_active"] is not None
+    for rows in listings[1:]:
+        assert rows[0]["id"] == "good"
+
+
 def test_writers_never_persist_an_out_of_window_timestamp(tmp_path):
     db = SessionDB(db_path=tmp_path / "state.db")
     try: