fix(state): report the never-active prune's exact skipped count

prune_never_active_keyed_sessions now returns (deleted, entries_dropped,
skipped) from its own guarded-row list. `sessions prune --never-active`
printed candidates-minus-deleted, which lumped rows that vanished
concurrently in with live-turn skips and needed an "or already gone"
hedge; it now reports only rows spared by a live turn or compression lock.

The docstring claimed a failed sweep rolls back whole; the row sweep is one
transaction but routing entries are dropped in a second write after it
commits, so say that.

The kept guard test now also covers a candidate that vanishes between
listing and the sweep: its routing entry must still be dropped (the r1
per-row loop left it behind, so the gateway would resume a nonexistent id).

hermes_cli/sessions_cmd.py

@@ -671,12 +671,12 @@ def _prune_never_active_keyed(db, args):
     if not args.yes and not _confirm_prompt(f"Delete {len(candidates)} session(s)? [y/N] "):
         print("Aborted.")
         return
-    deleted, routing_deleted = db.prune_never_active_keyed_sessions(
+    deleted, routing_deleted, skipped = db.prune_never_active_keyed_sessions(
         older_than_days=days, sessions_dir=_sessions_dir()
     )
     print(f"Deleted {deleted} never-active session(s) and {routing_deleted} stale routing entr(ies).")
-    if deleted < len(candidates):
-        print(f"Skipped {len(candidates) - deleted} session(s): mid-turn, or already gone.")
+    if skipped:
+        print(f"Skipped {skipped} session(s) with a live turn or compression lock.")
 
 
 def _note_pinned_skipped(db, filters, action):

hermes_state_gateway.py

@@ -397,9 +397,9 @@ class SessionGatewayMixin:
         return len(doomed)
 
     def prune_never_active_keyed_sessions(
-        self, *, older_than_days: float, sessions_dir: Optional[Path] = None) -> Tuple[int, int]:
+        self, *, older_than_days: float, sessions_dir: Optional[Path] = None) -> Tuple[int, int, int]:
         """Delete never-active keyed rows and the routing entries naming them; returns
-        ``(sessions_deleted, routing_entries_deleted)``. Deletion goes through
+        ``(sessions_deleted, routing_entries_deleted, sessions_skipped)``. Deletion goes through
         :meth:`delete_sessions` (one transaction; delegate cascade, FTS, transcripts).
         Rows under a live turn lease or compression lock are skipped (#123583): the
         never-active predicate reads committed state, so a keyed row whose first turn lease
@@ -407,16 +407,17 @@ class SessionGatewayMixin:
         mid-turn. Routing entries are dropped only after the rows commit, for every
         candidate except the guarded ones: rows that vanished concurrently are gone too,
         and a stale entry outliving its target would have the gateway resume a
-        nonexistent id. A failed sweep rolls back whole, so no entry is stranded."""
+        nonexistent id. The row sweep is one transaction; routing entries are dropped in a
+        second write after it commits. ``sessions_skipped`` counts only the guarded rows."""
         candidates = self.list_never_active_keyed_sessions(older_than_days=older_than_days)
         if not candidates:
-            return (0, 0)
+            return (0, 0, 0)
         ids = {str(row["id"]) for row in candidates}
         skipped: List[str] = []
         deleted = self.delete_sessions(
             list(ids), sessions_dir=sessions_dir, exclude_active_write_guards=True, skipped_ids=skipped)
         routing_deleted = self._delete_routing_entries_for_sessions(ids - set(skipped))
-        return (deleted, routing_deleted)
+        return (deleted, routing_deleted, len(skipped))
 
     def list_gateway_sessions(
         self, *, platform: Optional[str] = None, active_only: bool = True) -> List[Dict[str, Any]]:

tests/hermes_state/test_never_active_keyed_prune.py

@@ -110,7 +110,7 @@ class TestPrune:
         _insert(db, "keeper-young", age_days=1)
         _insert(db, "keeper-used", age_days=45, message_count=3)
 
-        deleted, _ = db.prune_never_active_keyed_sessions(older_than_days=30)
+        deleted, _, _ = db.prune_never_active_keyed_sessions(older_than_days=30)
 
         assert deleted == 2
         surviving = {
@@ -134,11 +134,9 @@ class TestPrune:
             scope="/home/dev/project",
         )
 
-        deleted, routing_deleted = db.prune_never_active_keyed_sessions(
-            older_than_days=30
-        )
+        result = db.prune_never_active_keyed_sessions(older_than_days=30)
 
-        assert (deleted, routing_deleted) == (1, 1)
+        assert result == (1, 1, 0)
         remaining = db._conn.execute(
             "SELECT session_key FROM gateway_routing"
         ).fetchall()
@@ -146,7 +144,7 @@ class TestPrune:
 
     def test_no_candidates_is_a_no_op(self, db):
         _insert(db, "keeper", age_days=1)
-        assert db.prune_never_active_keyed_sessions(older_than_days=30) == (0, 0)
+        assert db.prune_never_active_keyed_sessions(older_than_days=30) == (0, 0, 0)
         assert db._conn.execute("SELECT COUNT(*) FROM sessions").fetchone()[0] == 1
 
 
@@ -156,19 +154,31 @@ class TestPruneSkipsLiveTurns:
     empty. The prune must skip it (and keep its routing entry) instead of deleting it
     mid-turn."""
 
-    def test_guarded_row_survives_and_keeps_its_routing_entry(self, db):
+    def test_guarded_row_survives_and_keeps_its_routing_entry(self, db, monkeypatch):
         _insert(db, "live-junk", age_days=45)
         _insert(db, "dead-junk", age_days=45)
-        db.save_gateway_routing_entry(
-            "agent:main:telegram:dm:live-junk",
-            json.dumps({"session_id": "live-junk"}),
-            scope="/tmp/pytest-live",
-        )
+        _insert(db, "gone-junk", age_days=45)
+        for sid, scope in (("live-junk", "/tmp/pytest-live"), ("gone-junk", "/tmp/pytest-gone")):
+            db.save_gateway_routing_entry(
+                f"agent:main:telegram:dm:{sid}", json.dumps({"session_id": sid}), scope=scope)
         holder = f"pid={os.getpid()}:turn=1"
         assert db.try_acquire_session_turn_lease("live-junk", holder, ttl_seconds=300.0) is True
 
-        # only dead-junk (no routing entry); the guarded row is skipped, not fatal
-        assert db.prune_never_active_keyed_sessions(older_than_days=30) == (1, 0)
+        # gone-junk vanishes between listing and the sweep (a concurrent delete): it is not
+        # counted as deleted or skipped, but its routing entry must still go — a stale entry
+        # would have the gateway resume a nonexistent id.
+        real_list = db.list_never_active_keyed_sessions
+
+        def list_then_vanish(**kw):
+            rows = real_list(**kw)
+            db._conn.execute("DELETE FROM sessions WHERE id = 'gone-junk'")
+            db._conn.commit()
+            return rows
+
+        monkeypatch.setattr(db, "list_never_active_keyed_sessions", list_then_vanish)
+        # dead-junk deleted; the guarded row is skipped (counted), not fatal
+        assert db.prune_never_active_keyed_sessions(older_than_days=30) == (1, 1, 1)
+        monkeypatch.undo()
         assert db.get_session("live-junk") is not None
         remaining = {
             r[0]
@@ -177,5 +187,5 @@ class TestPruneSkipsLiveTurns:
         assert remaining == {"agent:main:telegram:dm:live-junk"}
 
         db.release_session_turn_lease("live-junk", holder)
-        assert db.prune_never_active_keyed_sessions(older_than_days=30) == (1, 1)
+        assert db.prune_never_active_keyed_sessions(older_than_days=30) == (1, 1, 0)
         assert db.get_session("live-junk") is None