refactor(sessions): make whole_lineages an explicit prune keyword

_prune_where popped whole_lineages out of **filters and prune_sessions
injected it by mutating that dict, so a filter-shaped kwarg silently
changed the SQL shape and a typo'd flag would be ignored rather than
rejected. It is now a keyword-only argument of _prune_where and the
list/count prune helpers; prune_sessions passes whole_lineages=True and
the CLI and dashboard previews pass it explicitly. Behaviour unchanged.

hermes_cli/sessions_cmd.py

@@ -639,7 +639,8 @@ def _note_pinned_skipped(db, filters, action):
     """Tell the user how many pinned rows bulk prune/archive spared (pin = durable keep; only
     `prune --include-pinned` opts in, archive always spares them)."""
     _base = {k: v for k, v in filters.items() if k != "include_pinned"}
-    with_pinned, without = (int(db.count_prune_matches(**_base, include_pinned=flag)) for flag in (True, False))
+    with_pinned, without = (int(db.count_prune_matches(**_base, include_pinned=flag, whole_lineages=action == "prune"))
+                            for flag in (True, False))
     skipped = max(with_pinned - without, 0)
     if not skipped:
         return
@@ -679,14 +680,13 @@ def _cmd_prune_or_archive(db, args, action):
     # Archive flips a compression lineage as a unit, matched through its tip (an old ancestor alone
     # never qualifies); the preview must show the same rows the archive will touch.
     filters["lineage_tips_only"] = not prune
-    # Prune deletes a compression lineage only as a unit; the preview must list the rows it deletes.
-    filters["whole_lineages"] = prune
     if not filters["include_pinned"]:
         _note_pinned_skipped(db, filters, action)
-    candidates = db.list_prune_candidates(**filters)
+    # Prune deletes a compression lineage only as a unit; the preview must list the rows it deletes.
+    candidates = db.list_prune_candidates(**filters, whole_lineages=prune)
     # Archive expands each matched tip to its compression lineage, so a direct-open count would
     # misdescribe its effect.
-    skipped_open = db.count_open_prune_matches(**filters) if prune else 0
+    skipped_open = db.count_open_prune_matches(**filters, whole_lineages=True) if prune else 0
     if skipped_open:
         print(f"Note: {skipped_open} open session{'' if skipped_open == 1 else 's'} also match these filters but "
               "will be skipped because prune only deletes ended sessions. Use `hermes sessions delete <id>` "

hermes_cli/web_routers/sessions.py

@@ -92,12 +92,13 @@ def _prune_sessions(body: SessionPrune):
         filters = {
             "older_than_days": effective_older_than, "started_before": body.started_before,
             "started_after": body.started_after,
-            "archived": None if body.include_archived else False, "whole_lineages": True,
+            "archived": None if body.include_archived else False,
             **{f: (getattr(body, f) or None) for f in _PRUNE_STR_FILTERS},
             **{f: getattr(body, f) for f in _PRUNE_NUM_FILTERS}}
-        skipped_open = db.count_open_prune_matches(**filters)
+        # Same whole-lineage selection prune_sessions applies, so the preview lists what it deletes.
+        skipped_open = db.count_open_prune_matches(**filters, whole_lineages=True)
         if body.dry_run:
-            rows = db.list_prune_candidates(**filters)
+            rows = db.list_prune_candidates(**filters, whole_lineages=True)
             return {
                 "ok": True,
                 "removed": 0,

hermes_state_maintenance.py

@@ -220,11 +220,10 @@ class SessionMaintenanceMixin:
             clauses.append("COALESCE(s.pinned, 0) = 0")
         return " AND ".join(clauses), params
 
-    def _prune_where(self, older_than_days, source, filters) -> Tuple[str, list]:
+    def _prune_where(self, older_than_days, source, filters, *, whole_lineages: bool = False) -> Tuple[str, list]:
         """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:
@@ -239,11 +238,11 @@ class SessionMaintenanceMixin:
         # 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]]:
+    def list_prune_candidates(self, older_than_days: Optional[float] = None, source: str = None, *,
+                              whole_lineages: bool = False, **filters) -> List[Dict[str, Any]]:
         """Dry-run: sessions a matching prune/archive would touch, oldest first (``older_than_days``
         = inactivity threshold: freshest of ``last_activity_at`` / latest message / ``started_at``)."""
-        where, params = self._prune_where(older_than_days, source, filters)
+        where, params = self._prune_where(older_than_days, source, filters, whole_lineages=whole_lineages)
         return [dict(row) for row in self._read_all(
             f"""SELECT s.id, s.source, s.title, s.model, s.started_at,
                            {_LAST_ACTIVE_SQL} AS last_active,
@@ -251,16 +250,16 @@ class SessionMaintenanceMixin:
                     FROM sessions s WHERE {where}
                     ORDER BY last_active ASC, s.started_at ASC""", params)]
 
-    def count_prune_matches(self, older_than_days: Optional[float] = None, source: str = None,
-                            **filters) -> int:
+    def count_prune_matches(self, older_than_days: Optional[float] = None, source: str = None, *,
+                            whole_lineages: bool = False, **filters) -> int:
         """Count-only :meth:`list_prune_candidates` (CLI reports spared pinned sessions)."""
-        where, params = self._prune_where(older_than_days, source, filters)
+        where, params = self._prune_where(older_than_days, source, filters, whole_lineages=whole_lineages)
         return int(self._read_one(f"SELECT COUNT(*) FROM sessions s WHERE {where}", params)[0])
 
-    def count_open_prune_matches(self, older_than_days: Optional[float] = None, source: str = None,
-                                 **filters) -> int:
+    def count_open_prune_matches(self, older_than_days: Optional[float] = None, source: str = None, *,
+                                 whole_lineages: bool = False, **filters) -> int:
         """Count open sessions a matching prune skips (``ended_at`` guard inverted); visibility-only."""
-        where, params = self._prune_where(older_than_days, source, filters)
+        where, params = self._prune_where(older_than_days, source, filters, whole_lineages=whole_lineages)
         ended_guard = "s.ended_at IS NOT NULL"
         if not where.startswith(ended_guard):
             raise RuntimeError("prune filter lost its ended-session safety guard")
@@ -303,8 +302,7 @@ class SessionMaintenanceMixin:
         ``exclude_active_write_guards`` (automatic maintenance) skips rows under a live turn lease
         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)
+        where, where_params = self._prune_where(older_than_days, source, filters, whole_lineages=True)
         removed_ids: list[str] = []
         def _do(conn):
             cursor = conn.execute(f"SELECT s.id FROM sessions s WHERE {where}", where_params)