fix(sessions): preserve recently active sessions during pruning

gateway/run.py

@@ -3592,8 +3592,8 @@ def __init__(self, config: Optional[GatewayConfig] = None):
             # hermes_state.get_last_init_error() for slash-command error strings.
             logger.warning("SQLite session store not available: %s", e)
 
-        # Opportunistic state.db maintenance: prune ended sessions older
-        # than sessions.retention_days + optional VACUUM. Tracks last-run
+        # Opportunistic state.db maintenance: prune ended sessions inactive
+        # for sessions.retention_days + optional VACUUM. Tracks last-run
         # in state_meta so it only actually executes once per
         # sessions.min_interval_hours.  Gateway is long-lived so blocking
         # a few seconds once per day is acceptable; failures are logged

hermes_cli/config.py

@@ -3217,14 +3217,15 @@ def _ensure_hermes_home_managed(home: Path):
     # reports 384MB+ databases with 68K+ messages, which slows down FTS5
     # inserts, /resume listing, and insights queries.
     "sessions": {
-        # When true, prune ended sessions older than retention_days once
+        # When true, prune ended sessions inactive for retention_days once
         # per (roughly) min_interval_hours at CLI/gateway/cron startup.
-        # Only touches ended sessions — active sessions are always preserved.
+        # Activity is the latest message timestamp, falling back to creation
+        # time for empty sessions. Active sessions are always preserved.
         # Default false: session history is valuable for search recall, and
         # silently deleting it could surprise users.  Opt in explicitly.
         "auto_prune": False,
-        # How many days of ended-session history to keep.  Matches the
-        # default of ``hermes sessions prune``.
+        # How many inactive days of ended-session history to keep. Matches
+        # the default of ``hermes sessions prune``.
         "retention_days": 90,
         # When true, auto-archive (soft-hide, never delete) sessions that
         # haven't been touched in ``auto_archive_days`` days, once per

hermes_cli/main.py

@@ -15127,7 +15127,7 @@ def _add_session_filter_args(p, default_older_help):
         p.add_argument(
             "--newer-than",
             metavar="AGE",
-            help="Only match sessions started within the last AGE "
+            help="Only match sessions active within the last AGE "
             "(e.g. '5h', '2d') or after an ISO timestamp",
         )
         p.add_argument(
@@ -16030,12 +16030,14 @@ def _export_one(session_id: str):
                 print(f"No sessions match ({describe_filters(filters)}).")
                 return
 
-            # Candidates are ordered oldest-first — surface the age span so
-            # the confirmation makes the blast radius obvious.
-            _oldest = candidates[0].get("started_at")
-            _newest = candidates[-1].get("started_at")
+            # Candidates are ordered by activity oldest-first. Surface that
+            # span so a long-lived but recently used conversation cannot look
+            # old merely because of its creation date.
+            _oldest = candidates[0].get("last_active")
+            _newest = candidates[-1].get("last_active")
             _span = (
-                f"oldest {format_epoch(_oldest)}, newest {format_epoch(_newest)}"
+                f"oldest activity {format_epoch(_oldest)}, "
+                f"newest activity {format_epoch(_newest)}"
             )
 
             if args.dry_run or not args.yes:
@@ -16048,7 +16050,7 @@ def _export_one(session_id: str):
                     title = (s.get("title") or "")[:36]
                     model = (s.get("model") or "-").split("/")[-1][:24]
                     print(
-                        f"  {s['id']}  {format_epoch(s['started_at']):<17} "
+                        f"  {s['id']}  {format_epoch(s.get('last_active')):<17} "
                         f"{s['source']:<10} {model:<24} "
                         f"{s['message_count']:>4} msgs  {title}"
                     )

hermes_cli/session_filters.py

@@ -86,13 +86,15 @@ def build_prune_filters(args: Any) -> Dict[str, Any]:
     ``--after``, ``--source``, ``--title``, ``--end-reason``, ``--cwd``,
     ``--min-messages``, ``--max-messages``, ``--archived``/``--no-archived``.
 
-    ``--before``/``--older-than`` both set the upper bound (started_before);
-    ``--after``/``--newer-than`` both set the lower bound (started_after).
-    When both a duration flag and an absolute flag target the same bound,
-    the tighter (more restrictive) bound wins.
+    ``--older-than`` / ``--newer-than`` bound last activity, while
+    ``--before`` / ``--after`` explicitly bound session start time. Last
+    activity is the latest message timestamp, falling back to ``started_at``
+    for empty sessions.
 
     Raises ``ValueError`` on unparseable values or an empty/inverted window.
     """
+    last_active_before: Optional[float] = None
+    last_active_after: Optional[float] = None
     started_before: Optional[float] = None
     started_after: Optional[float] = None
 
@@ -103,19 +105,23 @@ def _tighter(current: Optional[float], new: float, upper: bool) -> float:
 
     older_than = getattr(args, "older_than", None)
     if older_than is not None:
-        started_before = _tighter(
-            started_before, parse_point_in_time(older_than, "--older-than"), True
+        last_active_before = _tighter(
+            last_active_before,
+            parse_point_in_time(older_than, "--older-than"),
+            True,
+        )
+    newer_than = getattr(args, "newer_than", None)
+    if newer_than is not None:
+        last_active_after = _tighter(
+            last_active_after,
+            parse_point_in_time(newer_than, "--newer-than"),
+            False,
         )
     before = getattr(args, "before", None)
     if before is not None:
         started_before = _tighter(
             started_before, parse_point_in_time(before, "--before"), True
         )
-    newer_than = getattr(args, "newer_than", None)
-    if newer_than is not None:
-        started_after = _tighter(
-            started_after, parse_point_in_time(newer_than, "--newer-than"), False
-        )
     after = getattr(args, "after", None)
     if after is not None:
         started_after = _tighter(
@@ -128,16 +134,28 @@ def _tighter(current: Optional[float], new: float, upper: bool) -> float:
         and started_after >= started_before
     ):
         raise ValueError(
-            "Empty time window: the --after/--newer-than bound "
+            "Empty start-time window: the --after bound "
             f"({format_epoch(started_after)}) is not earlier than the "
-            f"--before/--older-than bound ({format_epoch(started_before)})."
+            f"--before bound ({format_epoch(started_before)})."
+        )
+    if (
+        last_active_before is not None
+        and last_active_after is not None
+        and last_active_after >= last_active_before
+    ):
+        raise ValueError(
+            "Empty activity window: the --newer-than bound "
+            f"({format_epoch(last_active_after)}) is not earlier than the "
+            f"--older-than bound ({format_epoch(last_active_before)})."
         )
 
     filters: Dict[str, Any] = {
         # older_than_days=None: the epoch bounds above are the whole story.
         # Without this, prune_sessions' default 90-day cutoff would silently
         # cap an --after/--newer-than-only window.
         "older_than_days": None,
+        "last_active_before": last_active_before,
+        "last_active_after": last_active_after,
         "started_before": started_before,
         "started_after": started_after,
         "source": getattr(args, "source", None),
@@ -165,6 +183,14 @@ def _tighter(current: Optional[float], new: float, upper: bool) -> float:
 def describe_filters(filters: Dict[str, Any]) -> str:
     """Human-readable summary of active filters for confirmation prompts."""
     parts = []
+    if filters.get("last_active_before") is not None:
+        parts.append(
+            f"last active before {format_epoch(filters['last_active_before'])}"
+        )
+    if filters.get("last_active_after") is not None:
+        parts.append(
+            f"last active after {format_epoch(filters['last_active_after'])}"
+        )
     if filters.get("started_before") is not None:
         parts.append(f"started before {format_epoch(filters['started_before'])}")
     if filters.get("started_after") is not None:

hermes_cli/web_server.py

@@ -11810,16 +11810,23 @@ def _prune_sessions(body: SessionPrune):
                 "ok": True,
                 "removed": 0,
                 "matched": len(rows),
-                # Rows are ordered oldest-first.
-                "oldest_started_at": rows[0]["started_at"] if rows else None,
-                "newest_started_at": rows[-1]["started_at"] if rows else None,
+                # Rows are ordered by last activity, not creation time.
+                "oldest_last_active": rows[0]["last_active"] if rows else None,
+                "newest_last_active": rows[-1]["last_active"] if rows else None,
+                "oldest_started_at": (
+                    min(r["started_at"] for r in rows) if rows else None
+                ),
+                "newest_started_at": (
+                    max(r["started_at"] for r in rows) if rows else None
+                ),
                 "sessions": [
                     {
                         "id": r["id"],
                         "source": r["source"],
                         "title": r.get("title"),
                         "model": r.get("model"),
                         "started_at": r["started_at"],
+                        "last_active": r["last_active"],
                         "message_count": r["message_count"],
                     }
                     for r in rows

hermes_state.py

@@ -8981,6 +8981,8 @@ def _do(conn):
     @staticmethod
     def _prune_filter_where(
         *,
+        last_active_before: Optional[float] = None,
+        last_active_after: Optional[float] = None,
         started_before: Optional[float] = None,
         started_after: Optional[float] = None,
         source: Optional[str] = None,
@@ -9023,6 +9025,24 @@ def _prune_filter_where(
         """
         clauses = ["s.ended_at IS NOT NULL"]
         params: list = []
+        if last_active_before is not None:
+            clauses.append(
+                """COALESCE(
+                       (SELECT MAX(m.timestamp) FROM messages m
+                        WHERE m.session_id = s.id),
+                       s.started_at
+                   ) < ?"""
+            )
+            params.append(last_active_before)
+        if last_active_after is not None:
+            clauses.append(
+                """COALESCE(
+                       (SELECT MAX(m.timestamp) FROM messages m
+                        WHERE m.session_id = s.id),
+                       s.started_at
+                   ) >= ?"""
+            )
+            params.append(last_active_after)
         if started_before is not None:
             clauses.append("s.started_at < ?")
             params.append(started_before)
@@ -9110,18 +9130,31 @@ def list_prune_candidates(
         Backs ``--dry-run`` and pre-confirmation counts. Accepts the same
         keyword filters as :meth:`_prune_filter_where` (unknown names raise
         ``TypeError`` there). Rows are ordered oldest-first and carry
-        ``id, source, title, model, started_at, ended_at, message_count,
-        archived``.
+        ``id, source, title, model, started_at, last_active, ended_at,
+        message_count, archived``. ``older_than_days`` is an inactivity
+        threshold: it uses the latest message timestamp, falling back to
+        ``started_at`` for sessions without messages.
         """
-        if filters.get("started_before") is None and older_than_days is not None:
-            filters["started_before"] = time.time() - (older_than_days * 86400)
+        if (
+            filters.get("last_active_before") is None
+            and filters.get("started_before") is None
+            and older_than_days is not None
+        ):
+            filters["last_active_before"] = time.time() - (
+                older_than_days * 86400
+            )
         where, params = self._prune_filter_where(source=source, **filters)
         with self._lock:
             cursor = self._conn.execute(
                 f"""SELECT s.id, s.source, s.title, s.model, s.started_at,
+                           COALESCE(
+                               (SELECT MAX(m.timestamp) FROM messages m
+                                WHERE m.session_id = s.id),
+                               s.started_at
+                           ) AS last_active,
                            s.ended_at, s.message_count, s.archived
                     FROM sessions s WHERE {where}
-                    ORDER BY s.started_at ASC""",
+                    ORDER BY last_active ASC, s.started_at ASC""",
                 params,
             )
             return [dict(row) for row in cursor.fetchall()]
@@ -9160,8 +9193,8 @@ def archive_stale_sessions(
         "Touched" is the latest message timestamp (falling back to
         ``started_at``) — i.e. real recency, not creation time — so a session
         created long ago but active yesterday is spared, while an old
-        abandoned one (even a still-open one) is swept. This differs from
-        :meth:`archive_sessions`, which ages on ``started_at`` and only ended
+        abandoned one (even a still-open one) is swept. Unlike
+        :meth:`archive_sessions`, this method can also archive unended
         sessions.
 
         Guards:
@@ -9210,15 +9243,19 @@ def prune_sessions(
     ) -> int:
         """Delete sessions matching the filters. Returns count deleted.
 
-        Default behavior (no keyword filters) is unchanged: delete ended
-        sessions older than ``older_than_days`` days, optionally restricted
-        to ``source``. Additional keyword filters AND together — the full
-        set is defined by :meth:`_prune_filter_where`:
+        By default, delete ended sessions inactive for
+        ``older_than_days`` days, optionally restricted to ``source``.
+        Activity is the latest message timestamp, falling back to
+        ``started_at`` for sessions without messages. Additional keyword
+        filters AND together — the full set is defined by
+        :meth:`_prune_filter_where`:
 
+        * ``last_active_before`` / ``last_active_after`` — epoch bounds on
+          the latest message timestamp (falling back to ``started_at``).
         * ``started_before`` / ``started_after`` — epoch bounds on
-          ``started_at``. ``started_before`` overrides ``older_than_days``;
-          pass ``older_than_days=None`` for no upper age bound (e.g. when
-          only pruning a recent window via ``started_after``).
+          ``started_at``. An explicit ``started_before`` overrides the
+          default ``older_than_days`` inactivity cutoff; pass
+          ``older_than_days=None`` for no implicit upper age bound.
         * ``title_like`` / ``model_like`` / ``branch_like`` —
           case-insensitive substring matches.
         * ``end_reason`` / ``provider`` / ``user_id`` / ``chat_id`` /
@@ -9240,8 +9277,14 @@ def prune_sessions(
         ``request_dump_*``) for every pruned session, outside the DB
         transaction.
         """
-        if filters.get("started_before") is None and older_than_days is not None:
-            filters["started_before"] = time.time() - (older_than_days * 86400)
+        if (
+            filters.get("last_active_before") is None
+            and filters.get("started_before") is None
+            and older_than_days is not None
+        ):
+            filters["last_active_before"] = time.time() - (
+                older_than_days * 86400
+            )
         where, where_params = self._prune_filter_where(source=source, **filters)
         removed_ids: list[str] = []
 
@@ -9952,7 +9995,7 @@ def maybe_auto_prune_and_vacuum(
         vacuum: bool = True,
         sessions_dir: Optional[Path] = None,
     ) -> Dict[str, Any]:
-        """Idempotent auto-maintenance: prune old sessions + optional VACUUM.
+        """Idempotent auto-maintenance: prune inactive sessions + optional VACUUM.
 
         Records the last run timestamp in state_meta so subsequent calls
         within ``min_interval_hours`` no-op. Designed to be called once at
@@ -10006,7 +10049,7 @@ def maybe_auto_prune_and_vacuum(
 
             if pruned > 0:
                 logger.info(
-                    "state.db auto-maintenance: pruned %d session(s) older than %d days%s",
+                    "state.db auto-maintenance: pruned %d session(s) inactive for %d days%s",
                     pruned,
                     retention_days,
                     " + VACUUM" if result["vacuumed"] else "",

tests/hermes_cli/test_dashboard_admin_endpoints.py

@@ -740,6 +740,8 @@ def test_prune_attr_filter_suppresses_default_cutoff(self):
         body = r.json()
         assert body["matched"] >= 1
         assert "oldest_started_at" in body and "newest_started_at" in body
+        assert "oldest_last_active" in body and "newest_last_active" in body
+        assert all("last_active" in session for session in body["sessions"])
 
     def test_prune_explicit_older_than_kept_with_attr_filter(self):
         # Explicit older_than_days is honored even alongside attribute filters.

tests/hermes_cli/test_session_filters.py

@@ -72,29 +72,38 @@ class TestBuildPruneFilters:
     def test_newer_than_sets_lower_bound_only(self):
         f = build_prune_filters(_ns(newer_than="5h"))
         assert f["started_before"] is None
-        assert f["started_after"] == pytest.approx(time.time() - 18000, abs=5)
+        assert f["started_after"] is None
+        assert f["last_active_after"] == pytest.approx(
+            time.time() - 18000, abs=5
+        )
         assert f["older_than_days"] is None  # no implicit 90d cap
 
     def test_older_than_bare_days(self):
         f = build_prune_filters(_ns(older_than="90"))
-        assert f["started_before"] == pytest.approx(
+        assert f["last_active_before"] == pytest.approx(
             time.time() - 90 * 86400, abs=5
         )
+        assert f["started_before"] is None
         assert f["started_after"] is None
 
     def test_window_before_and_after(self):
         f = build_prune_filters(_ns(after="10h", before="2h"))
         assert f["started_after"] < f["started_before"]
 
     def test_inverted_window_rejected(self):
-        with pytest.raises(ValueError, match="Empty time window"):
+        with pytest.raises(ValueError, match="Empty start-time window"):
             build_prune_filters(_ns(after="2h", before="10h"))
 
-    def test_tighter_bound_wins(self):
-        # --older-than 1d and --before 5h both set the upper bound;
-        # 1d ago is earlier (tighter for "older than") so it wins.
+    def test_inverted_activity_window_rejected(self):
+        with pytest.raises(ValueError, match="Empty activity window"):
+            build_prune_filters(_ns(newer_than="2h", older_than="10h"))
+
+    def test_activity_and_start_bounds_are_independent(self):
         f = build_prune_filters(_ns(older_than="1d", before="5h"))
         assert f["started_before"] == pytest.approx(
+            time.time() - 5 * 3600, abs=5
+        )
+        assert f["last_active_before"] == pytest.approx(
             time.time() - 86400, abs=5
         )
 
@@ -141,7 +150,7 @@ def test_describe_filters_extended(self):
     def test_describe_filters_mentions_active_parts(self):
         f = build_prune_filters(_ns(newer_than="5h", source="cli"))
         desc = describe_filters(f)
-        assert "started after" in desc
+        assert "last active after" in desc
         assert "source 'cli'" in desc
 
     def test_describe_filters_empty(self):