fix(sessions): don't let the empty-session sweep delete an archived transcript

`count_empty_sessions` / `delete_empty_sessions` — the dashboard's
"Delete empty (N)" affordance — defined "empty" as `sessions.message_count
= 0`. That column is a denormalized counter over the LIVE (`active = 1`)
rows only, and two production transcript-rewrite paths reset it on purpose
while keeping every dropped turn on disk as `active = 0`:

  * `replace_messages(..., archive_dropped=True)` — the rewind / edit /
    regenerate mode added in #82756 so a taken-back turn stays recoverable.
  * `archive_and_compact` — in-place compaction, which archives the
    pre-compaction transcript under the same session id (#38763).

A chat rewound to its first turn, or compacted with an empty live set,
therefore reports `message_count = 0` while still holding its entire
history — and those soft-archived rows are the only copy. A gateway reload
is what makes the row eligible: it stamps `ended_at` on every detached
session (`end_reason='ws_orphan_reap'`), satisfying the sweep's
`ended_at IS NOT NULL` gate. The next sweep then hard-deleted the session
row AND `DELETE FROM messages`, destroying the transcript silently.

Every other emptiness test in `hermes_state` already defends the counter
with a real `EXISTS (SELECT 1 FROM messages ...)` probe
(`delete_session_if_empty`, `prune_empty_ghost_sessions`,
`list_never_active_keyed_sessions`, `find_recoverable_session`). This
sweep was the only destructive path that trusted the counter alone. It now
uses the same probe, via one `_EMPTY_SESSION_WHERE` selector shared by the
count and the delete so the button's N and the sweep it triggers can never
disagree again. The counter stays as a cheap prefilter; `EXISTS` is the
authority.

Genuinely message-less rows are still swept — the feature is unchanged for
the case it was built for.

Fixes #95868

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011zTDHnWcBvQsJSX1fBLhuv

hermes_cli/web_routers/sessions.py

@@ -492,11 +492,16 @@ async def count_empty_sessions_endpoint(profile: Optional[str] = None):
 
 @manage_router.delete("/api/sessions/empty")
 async def delete_empty_sessions_endpoint(profile: Optional[str] = None):
-    """Delete every empty (``message_count == 0``), ended,
-    non-archived session in a single transaction.
+    """Delete every empty, ended, non-archived session in a single
+    transaction.
 
     Safety contract mirrors :meth:`SessionDB.delete_empty_sessions`:
 
+    * "Empty" means the session owns no rows in ``messages`` at all — not
+      merely ``message_count == 0``. A rewound or in-place-compacted chat
+      keeps its dropped turns as soft-archived (``active = 0``) rows while
+      the counter reads zero, and those rows are the only recoverable copy
+      of the transcript (#95868).
     * Active sessions are skipped (``ended_at IS NULL``) so a live
       agent isn't yanked mid-handshake.
     * Archived sessions are skipped — the user explicitly chose to

hermes_state.py

@@ -13173,10 +13173,40 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
             self._remove_session_files(sessions_dir, sid)
         return count
 
+    #: Selector shared by :meth:`count_empty_sessions` and
+    #: :meth:`delete_empty_sessions` so the button's count and the sweep it
+    #: triggers can never disagree about what "empty" means.
+    #:
+    #: ``message_count`` alone is NOT a safe emptiness test. It is a
+    #: denormalized counter that tracks the LIVE (``active = 1``) row set, and
+    #: two production transcript-rewrite paths deliberately reset it while
+    #: keeping the dropped turns on disk as ``active = 0``:
+    #: :meth:`replace_messages` with ``archive_dropped=True`` (the rewind /
+    #: edit / regenerate mode from #82756) and :meth:`archive_and_compact`
+    #: (in-place compaction). A chat rewound to its first turn, or compacted
+    #: with an empty live set, therefore reports ``message_count = 0`` while
+    #: still holding its entire recoverable transcript — and those
+    #: soft-archived rows are the ONLY copy, which is the whole point of
+    #: archiving instead of deleting. Deleting such a row destroys it silently.
+    #:
+    #: So the counter stays as a cheap prefilter and a real ``EXISTS`` probe is
+    #: the authority, exactly as every other emptiness test in this module
+    #: already does it (:meth:`delete_session_if_empty`,
+    #: :meth:`prune_empty_ghost_sessions`,
+    #: :meth:`list_never_active_keyed_sessions`). (#95868)
+    _EMPTY_SESSION_WHERE = (
+        "message_count = 0 "
+        "AND ended_at IS NOT NULL "
+        "AND archived = 0 "
+        "AND NOT EXISTS ("
+        "SELECT 1 FROM messages WHERE messages.session_id = sessions.id"
+        ")"
+    )
+
     def count_empty_sessions(self) -> int:
         """Return the count of empty, non-active, non-archived sessions.
 
-        "Empty" = ``message_count = 0`` AND the session has ended
+        "Empty" = the session holds no message rows at all AND has ended
         (``ended_at IS NOT NULL``) AND is not archived. The ``ended_at``
         guard matches the safety contract used by :meth:`prune_sessions`:
         only ended sessions are candidates for bulk deletion, so a freshly
@@ -13184,6 +13214,11 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         held open by the live agent — is never sniped out from under
         the runtime.
 
+        Emptiness is decided by :data:`_EMPTY_SESSION_WHERE`, which probes the
+        ``messages`` table instead of trusting the ``message_count`` counter —
+        see that constant for why the counter reads zero for rewound and
+        compacted sessions that still hold a full transcript.
+
         Backs the ``GET /api/sessions/empty/count`` endpoint that lets the
         web dashboard hide its "Delete empty" button when there's nothing
         to clean up, and pre-populate the confirm dialog with the actual
@@ -13191,10 +13226,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         """
         with self._lock:
             cursor = self._conn.execute(
-                "SELECT COUNT(*) FROM sessions "
-                "WHERE message_count = 0 "
-                "AND ended_at IS NOT NULL "
-                "AND archived = 0"
+                f"SELECT COUNT(*) FROM sessions WHERE {self._EMPTY_SESSION_WHERE}"
             )
             return cursor.fetchone()[0]
 
@@ -13206,9 +13238,11 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
 
         Mirrors :meth:`prune_sessions`' transactional shape:
 
-        * Selects candidate IDs first (``message_count = 0`` AND
-          ``ended_at IS NOT NULL`` AND ``archived = 0``) so we never
-          touch a live session or one the user deliberately archived.
+        * Selects candidate IDs first (:data:`_EMPTY_SESSION_WHERE`: no
+          message rows AND ``ended_at IS NOT NULL`` AND ``archived = 0``)
+          so we never touch a live session, one the user deliberately
+          archived, or one whose transcript survives only as soft-archived
+          (``active = 0``) rows after a rewind or an in-place compaction.
         * Orphans any child whose parent is in the kill list — children
           of an empty parent are kept and re-parented to ``NULL`` rather
           than cascade-deleted, matching ``delete_session`` /
@@ -13231,10 +13265,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
 
         def _do(conn):
             cursor = conn.execute(
-                "SELECT id FROM sessions "
-                "WHERE message_count = 0 "
-                "AND ended_at IS NOT NULL "
-                "AND archived = 0"
+                f"SELECT id FROM sessions WHERE {self._EMPTY_SESSION_WHERE}"
             )
             session_ids = {row["id"] for row in cursor.fetchall()}
 
@@ -13249,10 +13280,11 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
             )
 
             for sid in session_ids:
-                # DELETE FROM messages is paranoia — by construction
-                # these rows have ``message_count = 0`` — but if a
-                # bookkeeping bug ever lets the counter drift below the
-                # real row count, we still leave a clean FK state.
+                # DELETE FROM messages is paranoia — the selector's
+                # ``NOT EXISTS`` probe already proved these sessions own no
+                # message rows — but a row inserted between the SELECT and
+                # this statement would otherwise be left dangling, so we
+                # still leave a clean FK state.
                 conn.execute(
                     "DELETE FROM messages WHERE session_id = ?", (sid,)
                 )

tests/hermes_state/test_empty_sweep_archived_transcript_95868.py

@@ -0,0 +1,137 @@
+"""The empty-session sweep must not eat a rewound / compacted transcript (#95868).
+
+``count_empty_sessions`` / ``delete_empty_sessions`` back the dashboard's
+"Delete empty (N)" affordance. They used to define "empty" as
+``sessions.message_count = 0``, which is a denormalized counter over the LIVE
+(``active = 1``) rows only.
+
+Two production transcript-rewrite paths reset that counter on purpose while
+keeping every dropped turn on disk as ``active = 0``:
+
+* ``replace_messages(..., archive_dropped=True)`` — the rewind / edit /
+  regenerate mode added in #82756 precisely so a user can take a turn back
+  without the rows being unrecoverable.
+* ``archive_and_compact`` — in-place compaction, which archives the
+  pre-compaction transcript under the same session id (#38763).
+
+Rewind a chat back to its first turn, or compact it with an empty live set,
+and the row reports ``message_count = 0`` while still holding its entire
+recoverable history. Those soft-archived rows are the ONLY copy.
+
+A gateway reload is what makes such a row *eligible*: every detached session
+gets ``ended_at`` stamped (``end_reason='ws_orphan_reap'``), which satisfies
+the sweep's ``ended_at IS NOT NULL`` gate. The next "Delete empty" click then
+hard-deleted the session row AND ``DELETE FROM messages`` — silently, with no
+log line to trace it by.
+
+These tests pin the counter drift as real, and pin the sweep's refusal to act
+on it.
+"""
+
+import pytest
+
+from hermes_state import SessionDB
+
+
+@pytest.fixture()
+def db(tmp_path):
+    return SessionDB(db_path=tmp_path / "state.db")
+
+
+def _seed(db, session_id, turns=4):
+    """A populated, ended desktop chat — the shape #95868 lost."""
+    db.create_session(session_id, source="desktop", model="test-model")
+    for i in range(turns):
+        db.append_message(
+            session_id,
+            role="user" if i % 2 == 0 else "assistant",
+            content=f"turn {i}",
+        )
+    return session_id
+
+
+def _row_counts(db, session_id):
+    """(message_count column, real rows on disk) for *session_id*."""
+    with db._lock:
+        counter = db._conn.execute(
+            "SELECT message_count FROM sessions WHERE id = ?", (session_id,)
+        ).fetchone()["message_count"]
+        rows = db._conn.execute(
+            "SELECT COUNT(*) AS n FROM messages WHERE session_id = ?", (session_id,)
+        ).fetchone()["n"]
+    return counter, rows
+
+
+def test_rewind_to_empty_drifts_the_counter_to_zero(db):
+    """The premise: an archived-drop rewrite zeroes the counter, keeps the rows."""
+    _seed(db, "rewound")
+    assert _row_counts(db, "rewound") == (4, 4)
+
+    db.replace_messages("rewound", [], archive_dropped=True)
+
+    counter, rows = _row_counts(db, "rewound")
+    assert counter == 0, "message_count tracks the live set only"
+    assert rows == 4, "the dropped turns stay on disk as the recoverable copy"
+    assert len(db.get_messages("rewound", include_inactive=True)) == 4
+
+
+def test_sweep_spares_a_rewound_session(db):
+    """A rewound chat is not empty and must survive the sweep."""
+    _seed(db, "rewound")
+    db.replace_messages("rewound", [], archive_dropped=True)
+    # A gateway reload stamps ended_at on every detached session, which is what
+    # makes the row eligible for the sweep in the first place.
+    db.end_session("rewound", end_reason="ws_orphan_reap")
+
+    assert db.count_empty_sessions() == 0
+    assert db.delete_empty_sessions() == 0
+    assert db.get_session("rewound") is not None
+    assert len(db.get_messages("rewound", include_inactive=True)) == 4
+
+
+def test_sweep_spares_an_in_place_compacted_session(db):
+    """``archive_and_compact`` leaves the same zero-counter/live-rows shape."""
+    _seed(db, "compacted")
+    db.archive_and_compact("compacted", [])
+    assert _row_counts(db, "compacted") == (0, 4)
+
+    db.end_session("compacted", end_reason="ws_orphan_reap")
+
+    assert db.count_empty_sessions() == 0
+    assert db.delete_empty_sessions() == 0
+    assert db.get_session("compacted") is not None
+    assert len(db.get_messages("compacted", include_inactive=True)) == 4
+
+
+def test_genuinely_empty_sessions_are_still_swept(db):
+    """The fix must not neuter the feature: no rows at all is still empty."""
+    db.create_session("ghost", source="desktop")
+    db.end_session("ghost", end_reason="tui_close")
+
+    _seed(db, "populated")
+    db.end_session("populated", end_reason="tui_close")
+
+    assert db.count_empty_sessions() == 1
+    assert db.delete_empty_sessions() == 1
+    assert db.get_session("ghost") is None
+    assert db.get_session("populated") is not None
+
+
+def test_count_and_delete_agree_on_a_mixed_database(db):
+    """The button's N and the sweep it triggers must never disagree.
+
+    They read one shared selector; this pins that they still agree once
+    drifted-counter rows are in the mix.
+    """
+    db.create_session("ghost", source="desktop")
+    db.end_session("ghost", end_reason="tui_close")
+
+    _seed(db, "rewound")
+    db.replace_messages("rewound", [], archive_dropped=True)
+    db.end_session("rewound", end_reason="ws_orphan_reap")
+
+    _seed(db, "live")  # never ended — not a candidate either way
+
+    counted = db.count_empty_sessions()
+    assert counted == 1
+    assert db.delete_empty_sessions() == counted