fix(tui): sweep orphaned tui/desktop/subagent session rows at gateway startup

cli-config.yaml.example

@@ -866,6 +866,19 @@ max_concurrent_sessions: null
 # explicitly want one shared "room brain" per group/channel.
 group_sessions_per_user: true
 
+# Session storage maintenance (state.db).
+sessions:
+  # Close tui/desktop/subagent session rows orphaned by a dead gateway
+  # process. The normal disconnect cleanup runs on an in-process grace
+  # timer, so a gateway restart (update, crash, systemd) leaves those rows
+  # permanently "active". On every gateway boot — stdio TUI *and* the
+  # desktop/dashboard WS sidecar — rows whose start time AND newest message
+  # are both older than the session TTL (HERMES_TUI_SESSION_TTL_S, default
+  # 6h) are closed with end_reason "startup_orphan_reap". Messaging-platform
+  # sessions (Telegram, Discord, ...) are never touched; live in-memory
+  # sessions are excluded; swept sessions stay resumable.
+  orphan_reaper: true
+
 # ─────────────────────────────────────────────────────────────────────────────
 # API Server — per-client model routing
 # ─────────────────────────────────────────────────────────────────────────────

hermes_cli/config_defaults.py

@@ -3048,6 +3048,17 @@
         # the sweep on every CLI invocation).  Tracked via state_meta in
         # state.db itself, so it's shared across all processes.
         "min_interval_hours": 24,
+        # Close tui/desktop/subagent session rows orphaned by a dead gateway
+        # process (#65194).  The ws-orphan grace timer is in-process, so a
+        # gateway restart leaves disconnected sessions ``ended_at IS NULL``
+        # forever — phantom "active" rows in /resume and dashboards.  On
+        # every gateway boot (stdio TUI *and* the desktop/dashboard WS
+        # sidecar), rows whose start time AND newest message are both older
+        # than the session TTL (HERMES_TUI_SESSION_TTL_S, default 6h) are
+        # closed with end_reason='startup_orphan_reap'.  Messaging-gateway
+        # sessions (telegram, discord, ...) are never touched; swept
+        # sessions stay resumable.  Live in-memory sessions are excluded.
+        "orphan_reaper": True,
         # Legacy per-session JSON snapshot writer.  When true, the agent
         # rewrites ``~/.hermes/sessions/session_{sid}.json`` on every turn
         # boundary with the full message list.  state.db is canonical and

hermes_state.py

@@ -7888,6 +7888,73 @@ def _do(conn):
 
         return self._execute_write(_do) or 0
 
+    def sweep_orphaned_sessions(
+        self,
+        *,
+        max_idle_seconds: float,
+        sources: Tuple[str, ...] = ("tui", "desktop", "subagent"),
+        exclude_ids: Tuple[str, ...] = (),
+    ) -> List[str]:
+        """Close session rows orphaned by a dead gateway process (#65194).
+
+        The TUI/desktop gateway reaps disconnected websocket sessions with an
+        in-process ``threading.Timer`` grace timer; a gateway restart destroys
+        the timer and leaves the row ``ended_at IS NULL`` forever.  This is
+        the startup-time complement: it closes rows for the given ``sources``
+        whose ``started_at`` AND newest ``messages.timestamp`` are both older
+        than ``max_idle_seconds``, with a distinct
+        ``end_reason='startup_orphan_reap'`` for traceability.
+
+        Both timestamps must be stale on purpose: message recency alone would
+        sweep a freshly created compression/branch child carrying old copied
+        message timestamps, while ``started_at`` alone would sweep a
+        long-lived session that is still actively producing messages.
+        Message-less rows fall back to ``started_at`` via COALESCE.
+
+        Only pass sources owned by the local UI stack (never messaging-gateway
+        platforms like ``telegram`` — ending those triggers the #60609 routing
+        loop).  ``exclude_ids`` spares rows this process still holds in
+        memory (a ``session.resume`` that landed during the startup grace
+        window).  Non-destructive: messages are preserved and the row remains
+        resumable.  First-reason-wins is preserved via ``ended_at IS NULL``.
+
+        The SELECT + UPDATE run in one ``BEGIN IMMEDIATE`` write, so a sibling
+        process cannot sneak a new message or end-reason between the
+        staleness check and the close.  Returns the swept session ids.
+        """
+        srcs = tuple(s for s in sources if s)
+        if max_idle_seconds <= 0 or not srcs:
+            return []
+        cutoff = time.time() - max_idle_seconds
+        placeholders = ",".join("?" for _ in srcs)
+        staleness = (
+            "started_at < ? AND COALESCE((SELECT MAX(m.timestamp) FROM messages m"
+            " WHERE m.session_id = sessions.id), started_at) < ?"
+        )
+
+        def _do(conn):
+            rows = conn.execute(
+                f"SELECT id FROM sessions WHERE ended_at IS NULL"
+                f" AND source IN ({placeholders}) AND {staleness}",
+                (*srcs, cutoff, cutoff),
+            ).fetchall()
+            excluded = {str(x) for x in exclude_ids if x}
+            victims = [str(r["id"]) for r in rows if str(r["id"]) not in excluded]
+            if not victims:
+                return []
+            now = time.time()
+            marks = ",".join("?" for _ in victims)
+            # Re-apply the same staleness predicate under the write lock so a
+            # row that raced to activity between SELECT and UPDATE is spared.
+            conn.execute(
+                f"UPDATE sessions SET ended_at = ?, end_reason = 'startup_orphan_reap'"
+                f" WHERE id IN ({marks}) AND ended_at IS NULL AND {staleness}",
+                (now, *victims, cutoff, cutoff),
+            )
+            return victims
+
+        return self._execute_write(_do) or []
+
     def get_session(self, session_id: str) -> Optional[Dict[str, Any]]:
         """Get a session by ID."""
         # Cost/usage readers (/status, /usage, gateway endpoints) reach the

tests/hermes_state/test_sweep_orphaned_sessions.py

@@ -0,0 +1,173 @@
+"""Tests for #65194: startup-time sweep of orphaned UI-stack sessions.
+
+The TUI/desktop gateway reaps disconnected websocket sessions with an
+in-process ``threading.Timer`` grace timer.  A gateway restart destroys the
+timer, so the session row stays ``ended_at IS NULL`` forever — nothing
+re-checks stale rows on the next boot.  ``SessionDB.sweep_orphaned_sessions()``
+is the DB-level startup sweep that closes such rows with a distinct
+``end_reason='startup_orphan_reap'``.
+
+Staleness requires BOTH ``started_at`` and the newest ``messages.timestamp``
+to be older than the cutoff:
+
+* message-recency alone would sweep a freshly created compression/branch
+  child that carries old *copied* message timestamps;
+* ``started_at`` alone would sweep a long-lived session that is still
+  actively producing messages.
+"""
+
+import time
+
+import pytest
+
+from hermes_state import SessionDB
+
+IDLE_S = 6 * 3600  # mirror the TUI gateway's default session TTL
+
+
+@pytest.fixture
+def db(tmp_path):
+    return SessionDB(tmp_path / "state.db")
+
+
+def _backdate_session(db: SessionDB, session_id: str, ts: float) -> None:
+    db._conn.execute(
+        "UPDATE sessions SET started_at = ? WHERE id = ?", (ts, session_id)
+    )
+    db._conn.commit()
+
+
+def _set_message_timestamps(db: SessionDB, session_id: str, ts: float) -> None:
+    db._conn.execute(
+        "UPDATE messages SET timestamp = ? WHERE session_id = ?", (ts, session_id)
+    )
+    db._conn.commit()
+
+
+def _make_session(
+    db: SessionDB,
+    session_id: str,
+    *,
+    source: str,
+    started_at: float,
+    message_at: float = None,
+) -> None:
+    db.create_session(session_id, source=source)
+    if message_at is not None:
+        db.append_message(session_id, role="user", content="hello")
+        _set_message_timestamps(db, session_id, message_at)
+    _backdate_session(db, session_id, started_at)
+
+
+class TestSweepOrphanedSessions:
+    def test_stale_tui_session_swept(self, db):
+        stale = time.time() - 8 * 3600
+        _make_session(db, "stale-tui", source="tui", started_at=stale, message_at=stale)
+
+        assert db.sweep_orphaned_sessions(max_idle_seconds=IDLE_S) == ["stale-tui"]
+
+        row = db.get_session("stale-tui")
+        assert row["ended_at"] is not None
+        assert row["end_reason"] == "startup_orphan_reap"
+
+    def test_stale_desktop_session_swept(self, db):
+        """Desktop chat rows use the same gateway and the same Timer path."""
+        stale = time.time() - 8 * 3600
+        _make_session(
+            db, "stale-desktop", source="desktop", started_at=stale, message_at=stale
+        )
+
+        assert db.sweep_orphaned_sessions(max_idle_seconds=IDLE_S) == ["stale-desktop"]
+        assert db.get_session("stale-desktop")["end_reason"] == "startup_orphan_reap"
+
+    def test_stale_subagent_session_swept(self, db):
+        stale = time.time() - 8 * 3600
+        _make_session(
+            db, "stale-sub", source="subagent", started_at=stale, message_at=stale
+        )
+
+        assert db.sweep_orphaned_sessions(max_idle_seconds=IDLE_S) == ["stale-sub"]
+        assert db.get_session("stale-sub")["end_reason"] == "startup_orphan_reap"
+
+    def test_recent_message_spares_old_session(self, db):
+        """A long-lived session that is still talking is NOT an orphan."""
+        stale = time.time() - 48 * 3600
+        _make_session(
+            db, "active", source="tui", started_at=stale, message_at=time.time()
+        )
+
+        assert db.sweep_orphaned_sessions(max_idle_seconds=IDLE_S) == []
+        assert db.get_session("active")["ended_at"] is None
+
+    def test_fresh_session_with_old_copied_messages_spared(self, db):
+        """Compression/branch children copy history — old message timestamps
+        on a just-created row must not get it swept."""
+        stale = time.time() - 8 * 3600
+        _make_session(
+            db, "fresh-child", source="tui", started_at=time.time(), message_at=stale
+        )
+
+        assert db.sweep_orphaned_sessions(max_idle_seconds=IDLE_S) == []
+        assert db.get_session("fresh-child")["ended_at"] is None
+
+    def test_gateway_owned_source_not_swept(self, db):
+        """telegram/discord/... rows belong to the messaging gateway (#60609)."""
+        stale = time.time() - 8 * 3600
+        _make_session(
+            db, "tg-sess", source="telegram", started_at=stale, message_at=stale
+        )
+
+        assert db.sweep_orphaned_sessions(max_idle_seconds=IDLE_S) == []
+        assert db.get_session("tg-sess")["ended_at"] is None
+
+    def test_already_ended_session_untouched(self, db):
+        """First end_reason wins — the sweep never rewrites history."""
+        stale = time.time() - 8 * 3600
+        _make_session(db, "done", source="tui", started_at=stale, message_at=stale)
+        db.end_session("done", "user_exit")
+
+        assert db.sweep_orphaned_sessions(max_idle_seconds=IDLE_S) == []
+        assert db.get_session("done")["end_reason"] == "user_exit"
+
+    def test_stale_empty_session_swept_fresh_spared(self, db):
+        """Rows without messages fall back to started_at staleness."""
+        stale = time.time() - 8 * 3600
+        _make_session(db, "stale-empty", source="tui", started_at=stale)
+        _make_session(db, "fresh-empty", source="tui", started_at=time.time())
+
+        assert db.sweep_orphaned_sessions(max_idle_seconds=IDLE_S) == ["stale-empty"]
+        assert db.get_session("stale-empty")["end_reason"] == "startup_orphan_reap"
+        assert db.get_session("fresh-empty")["ended_at"] is None
+
+    def test_exclude_ids_spares_live_row(self, db):
+        """A row this process still holds in memory must not be closed."""
+        stale = time.time() - 8 * 3600
+        _make_session(db, "live-tui", source="tui", started_at=stale, message_at=stale)
+        _make_session(db, "dead-tui", source="tui", started_at=stale, message_at=stale)
+
+        swept = db.sweep_orphaned_sessions(
+            max_idle_seconds=IDLE_S, exclude_ids=("live-tui",)
+        )
+        assert swept == ["dead-tui"]
+        assert db.get_session("live-tui")["ended_at"] is None
+        assert db.get_session("dead-tui")["end_reason"] == "startup_orphan_reap"
+
+    def test_custom_sources_respected(self, db):
+        stale = time.time() - 8 * 3600
+        _make_session(db, "stale-cli", source="cli", started_at=stale, message_at=stale)
+        _make_session(db, "stale-tui", source="tui", started_at=stale, message_at=stale)
+
+        assert db.sweep_orphaned_sessions(
+            max_idle_seconds=IDLE_S, sources=("cli",)
+        ) == ["stale-cli"]
+        assert db.get_session("stale-cli")["end_reason"] == "startup_orphan_reap"
+        assert db.get_session("stale-tui")["ended_at"] is None
+
+    def test_returns_empty_on_empty_db(self, db):
+        assert db.sweep_orphaned_sessions(max_idle_seconds=IDLE_S) == []
+
+    def test_zero_ttl_is_noop(self, db):
+        stale = time.time() - 8 * 3600
+        _make_session(db, "stale-tui", source="tui", started_at=stale, message_at=stale)
+        assert db.sweep_orphaned_sessions(max_idle_seconds=0) == []
+        assert db.get_session("stale-tui")["ended_at"] is None