fix(sessions): prune keeps the compressed-away start of a chat still in use

Retention prune (the default-on startup auto-prune, `hermes sessions prune`
and the dashboard prune) aged every session row on its own. A conversation
that rotating compression split into segments has an ended, old root by
construction, so once that root passed retention_days it was deleted while
the conversation's live tip was still being written: the pre-compression
turns vanished from the resume/Desktop history and from session_search, and
the tip was orphaned.

Prune now deletes a compression ancestor only together with every
continuation after it (`whole_lineages`), so a lineage ages through its
newest segment and goes as a unit once the whole conversation qualifies.
Branch, delegate, reset and tool children do not count as continuations.
The CLI and dashboard prune previews pass the same flag, so they list what
prune deletes; bulk export keeps its current selection.

(cherry picked from commit b082fc6ffa02607cba219a5dfa361d4f3bf4c266)

hermes_state_maintenance.py

@@ -8,7 +8,8 @@ from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple
 
 from hermes_state_common import (
-    AUTO_VACUUM_MIN_FREELIST_RATIO, _id_chunks, _placeholders, _sql_session_last_active, escape_like as _escape_like
+    AUTO_VACUUM_MIN_FREELIST_RATIO, _id_chunks, _placeholders, _sql_json_extract, _sql_session_last_active,
+    escape_like as _escape_like
 )
 from hermes_startup_watchdog import report_startup_progress
 
@@ -74,6 +75,27 @@ _PRUNE_FILTERS = (
 )
 _PRUNE_FILTER_NAMES = frozenset(name for name, _, _ in _PRUNE_FILTERS) | {"archived", "include_pinned", "lineage_tips_only"}
 
+# Child ``c`` continues compression-ended ``p``; a fork names ``p`` in its marker (compression copies
+# ``model_config``, so a marker naming another row is inherited, not a fork of ``p``).
+_CONTINUATION_EDGE_SQL = " AND ".join([
+    "p.end_reason = 'compression'",
+    *(f"COALESCE({_sql_json_extract('c.model_config', f'$.{marker}')}, '') != p.id"
+      for marker in ("_branched_from", "_delegate_from", "_reset_from")),
+    "COALESCE(c.source, '') != 'tool'",
+])
+
+
+def _continued_ancestors_sql(candidates_where: str) -> str:
+    """Compression ancestors of every row *candidates_where* (alias ``s``) does not select."""
+    return ("WITH RECURSIVE kept(id) AS ("
+            " SELECT p.id FROM sessions c JOIN sessions p ON p.id = c.parent_session_id"
+            f" WHERE {_CONTINUATION_EDGE_SQL}"
+            f" AND NOT EXISTS (SELECT 1 FROM sessions s WHERE s.id = c.id AND {candidates_where})"
+            " UNION"
+            " SELECT p.id FROM kept k JOIN sessions c ON c.id = k.id JOIN sessions p ON p.id = c.parent_session_id"
+            f" WHERE {_CONTINUATION_EDGE_SQL}"
+            ") SELECT id FROM kept")
+
 
 class SessionMaintenanceMixin:
     """Retention pruning, stale-session archiving and VACUUM policy for SessionDB."""
@@ -205,7 +227,10 @@ class SessionMaintenanceMixin:
         return " AND ".join(clauses), params
 
     def _prune_where(self, older_than_days, source, filters) -> Tuple[str, list]:
-        """Translate the legacy age window into the shared activity filter, then build WHERE."""
+        """Translate the legacy age window into the shared activity filter, then build WHERE.
+        ``whole_lineages`` (prune) keeps a compression ancestor while any continuation after it
+        is unmatched."""
+        whole_lineages = filters.pop("whole_lineages", False)
         if (older_than_days is not None and filters.get("last_active_before") is None
                 and filters.get("started_before") is None):
             if older_than_days < 0:
@@ -213,7 +238,12 @@ class SessionMaintenanceMixin:
                     f"older_than_days must be >= 0, got {older_than_days!r}: a negative "
                     "retention builds a future cutoff that matches every ended session.")
             filters["last_active_before"] = time.time() - (older_than_days * 86400)
-        return self._prune_filter_where(source=source, **filters)
+        where, params = self._prune_filter_where(source=source, **filters)
+        if not whole_lineages:
+            return where, params
+        # A compressed-away segment ages with its conversation, not on its own: while any later
+        # segment stays, deleting it would cut the start off a chat that is still in use.
+        return f"{where} AND s.id NOT IN ({_continued_ancestors_sql(where)})", [*params, *params]
 
     def list_prune_candidates(self, older_than_days: Optional[float] = None, source: str = None,
                               **filters) -> List[Dict[str, Any]]:
@@ -277,7 +307,9 @@ class SessionMaintenanceMixin:
         Children outside the window are orphaned (parent NULLed), not cascade-deleted.  With
         *sessions_dir*, transcript files are removed outside the DB transaction.
         ``exclude_active_write_guards`` (automatic maintenance) skips rows under a live turn lease
-        or compression lock while expired/dead holders are reclaimed and fenced."""
+        or compression lock while expired/dead holders are reclaimed and fenced.  A compression
+        ancestor is deleted only together with every continuation after it (``whole_lineages``)."""
+        filters["whole_lineages"] = True
         where, where_params = self._prune_where(older_than_days, source, filters)
         removed_ids: list[str] = []
         def _do(conn):