From bc000c9f87d7a2599b38c1c9ed65cdb83575b2cd Mon Sep 17 00:00:00 2001 From: halaprix <6533433+halaprix@users.noreply.github.com> Date: Wed, 19 Aug 2026 09:41:31 +0200 Subject: [PATCH 1/2] fix(tui): sweep orphaned tui/desktop/subagent session rows at gateway startup Close session rows left ended_at IS NULL when the in-process websocket orphan timer dies with the process (#65194). Dual-clock staleness (started_at AND newest message), desktop included, live in-memory sessions excluded, scheduled once from both entry.main and the WS sidecar so desktop/dashboard boots also run the sweep. --- cli-config.yaml.example | 13 + hermes_cli/config_defaults.py | 11 + hermes_state.py | 67 +++++ .../test_sweep_orphaned_sessions.py | 173 ++++++++++++ .../tui_gateway/test_startup_orphan_sweep.py | 249 ++++++++++++++++++ tui_gateway/entry.py | 10 + tui_gateway/server.py | 110 ++++++++ tui_gateway/ws.py | 9 + website/docs/user-guide/configuration.md | 11 + 9 files changed, 653 insertions(+) create mode 100644 tests/hermes_state/test_sweep_orphaned_sessions.py create mode 100644 tests/tui_gateway/test_startup_orphan_sweep.py diff --git a/cli-config.yaml.example b/cli-config.yaml.example index d2228c76a74ba..39afeec2430f7 100644 --- a/cli-config.yaml.example +++ b/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 # ───────────────────────────────────────────────────────────────────────────── diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index 29ae25d7f700d..b4e85c9f24a7f 100644 --- a/hermes_cli/config_defaults.py +++ b/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 diff --git a/hermes_state.py b/hermes_state.py index e5df0c816a1f5..9f83b040f0887 100644 --- a/hermes_state.py +++ b/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 diff --git a/tests/hermes_state/test_sweep_orphaned_sessions.py b/tests/hermes_state/test_sweep_orphaned_sessions.py new file mode 100644 index 0000000000000..4554513f2f17e --- /dev/null +++ b/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 diff --git a/tests/tui_gateway/test_startup_orphan_sweep.py b/tests/tui_gateway/test_startup_orphan_sweep.py new file mode 100644 index 0000000000000..971c2129ffa88 --- /dev/null +++ b/tests/tui_gateway/test_startup_orphan_sweep.py @@ -0,0 +1,249 @@ +"""Tests for #65194: the gateway's startup-time orphaned-session sweep. + +A gateway restart destroys the in-process ws-orphan grace timers +(``_schedule_ws_orphan_reap``), so rows for sessions that died with the +previous process stay ``ended_at IS NULL`` forever. Both gateway entry +points — stdio ``entry.main()`` and the desktop/dashboard WS sidecar +``handle_ws`` — must schedule a DB-level sweep, gated by +``sessions.orphan_reaper`` (default on) and the gateway's session TTL, +without ever blocking or crashing startup. +""" + +from __future__ import annotations + +import io +import time +import types + +from hermes_state import SessionDB +from tui_gateway import entry, server + + +IDLE_S = 6 * 3600 + + +def _seed_session(db, session_id, *, source, last_active, started_at=None): + db.create_session(session_id, source=source) + db.append_message(session_id, role="user", content="hello") + with db._lock: + db._conn.execute( + "UPDATE sessions SET started_at = ? WHERE id = ?", + (last_active if started_at is None else started_at, session_id), + ) + db._conn.execute( + "UPDATE messages SET timestamp = ? WHERE session_id = ?", + (last_active, session_id), + ) + db._conn.commit() + + +class TestSweepOrphanedSessionRows: + def test_ends_stale_tui_desktop_and_subagent(self, monkeypatch, tmp_path): + db = SessionDB(tmp_path / "state.db") + stale = time.time() - 8 * 3600 + _seed_session(db, "stale-tui", source="tui", last_active=stale) + _seed_session(db, "stale-desktop", source="desktop", last_active=stale) + _seed_session(db, "stale-sub", source="subagent", last_active=stale) + monkeypatch.setattr(server, "_get_db", lambda: db) + monkeypatch.setattr(server, "_SESSION_TTL_S", float(IDLE_S)) + monkeypatch.setattr(server, "_sessions", {}) + + swept = server._sweep_orphaned_session_rows() + + assert sorted(swept) == ["stale-desktop", "stale-sub", "stale-tui"] + for sid in ("stale-tui", "stale-desktop", "stale-sub"): + row = db.get_session(sid) + assert row["ended_at"] is not None + assert row["end_reason"] == "startup_orphan_reap" + + def test_spares_fresh_row_with_old_copied_history(self, monkeypatch, tmp_path): + db = SessionDB(tmp_path / "state.db") + old_history = time.time() - 8 * 3600 + _seed_session( + db, + "fresh-branch", + source="tui", + last_active=old_history, + started_at=time.time(), + ) + monkeypatch.setattr(server, "_get_db", lambda: db) + monkeypatch.setattr(server, "_SESSION_TTL_S", float(IDLE_S)) + monkeypatch.setattr(server, "_sessions", {}) + + assert server._sweep_orphaned_session_rows() == [] + assert db.get_session("fresh-branch")["ended_at"] is None + + def test_spares_live_in_memory_and_gateway_rows(self, monkeypatch, tmp_path): + db = SessionDB(tmp_path / "state.db") + stale = time.time() - 8 * 3600 + _seed_session(db, "resumed-tui", source="tui", last_active=stale) + _seed_session(db, "gateway-row", source="telegram", last_active=stale) + _seed_session(db, "recent-tui", source="tui", last_active=time.time() - 30) + monkeypatch.setattr(server, "_get_db", lambda: db) + monkeypatch.setattr(server, "_SESSION_TTL_S", float(IDLE_S)) + monkeypatch.setattr( + server, + "_sessions", + { + "mem-sid": { + "agent": types.SimpleNamespace(session_id="resumed-tui"), + "session_key": "resumed-tui", + } + }, + ) + + assert server._sweep_orphaned_session_rows() == [] + for row_id in ("resumed-tui", "gateway-row", "recent-tui"): + assert db.get_session(row_id)["ended_at"] is None + + def test_leaves_already_ended_rows_untouched(self, monkeypatch, tmp_path): + db = SessionDB(tmp_path / "state.db") + stale = time.time() - 8 * 3600 + _seed_session(db, "reaped-tui", source="tui", last_active=stale) + db.end_session("reaped-tui", "ws_orphan_reap") + before = db.get_session("reaped-tui") + monkeypatch.setattr(server, "_get_db", lambda: db) + monkeypatch.setattr(server, "_SESSION_TTL_S", float(IDLE_S)) + monkeypatch.setattr(server, "_sessions", {}) + + assert server._sweep_orphaned_session_rows() == [] + after = db.get_session("reaped-tui") + assert after["end_reason"] == "ws_orphan_reap" + assert after["ended_at"] == before["ended_at"] + + def test_zero_ttl_skips_sweep(self, monkeypatch, tmp_path): + db = SessionDB(tmp_path / "state.db") + stale = time.time() - 8 * 3600 + _seed_session(db, "stale-tui", source="tui", last_active=stale) + monkeypatch.setattr(server, "_get_db", lambda: db) + monkeypatch.setattr(server, "_SESSION_TTL_S", 0.0) + monkeypatch.setattr(server, "_sessions", {}) + + assert server._sweep_orphaned_session_rows() == [] + assert db.get_session("stale-tui")["ended_at"] is None + + +class TestScheduleStartupOrphanSweep: + def test_once_per_process_and_config_and_ttl_gates(self, monkeypatch): + started = {"count": 0} + + class _Timer: + def __init__(self, *a, **k): + pass + + def start(self): + started["count"] += 1 + + monkeypatch.setattr(server.threading, "Timer", _Timer) + monkeypatch.setattr(server, "_WS_ORPHAN_REAP_GRACE_S", 20.0) + monkeypatch.setattr(server, "_SESSION_TTL_S", float(IDLE_S)) + + monkeypatch.setattr(server, "_startup_orphan_sweep_ran", False) + monkeypatch.setattr(server, "_session_orphan_reaper_enabled", lambda: False) + server._schedule_startup_orphan_sweep() + assert started["count"] == 0 + assert server._startup_orphan_sweep_ran is False + + monkeypatch.setattr(server, "_session_orphan_reaper_enabled", lambda: True) + server._schedule_startup_orphan_sweep() + server._schedule_startup_orphan_sweep() + assert started["count"] == 1 + assert server._startup_orphan_sweep_ran is True + + monkeypatch.setattr(server, "_startup_orphan_sweep_ran", False) + monkeypatch.setattr(server, "_WS_ORPHAN_REAP_GRACE_S", 0.0) + server._schedule_startup_orphan_sweep() + assert started["count"] == 1 + + monkeypatch.setattr(server, "_WS_ORPHAN_REAP_GRACE_S", 20.0) + monkeypatch.setattr(server, "_SESSION_TTL_S", 0.0) + monkeypatch.setattr(server, "_startup_orphan_sweep_ran", False) + server._schedule_startup_orphan_sweep() + assert started["count"] == 1 + + def test_config_flag_reads_sessions_orphan_reaper(self, monkeypatch): + monkeypatch.setattr( + server, "_load_cfg", lambda: {"sessions": {"orphan_reaper": False}} + ) + assert server._session_orphan_reaper_enabled() is False + + monkeypatch.setattr(server, "_load_cfg", lambda: {}) + assert server._session_orphan_reaper_enabled() is True + + monkeypatch.setattr(server, "_load_cfg", lambda: (_ for _ in ()).throw(RuntimeError("boom"))) + assert server._session_orphan_reaper_enabled() is True + + +class TestEntryAndWsWiring: + def test_main_schedules_sweep(self, monkeypatch): + scheduled = {"n": 0} + + def _schedule(): + scheduled["n"] += 1 + + monkeypatch.setattr(server, "_schedule_startup_orphan_sweep", _schedule) + monkeypatch.setattr(entry, "_install_sidecar_publisher", lambda: None) + monkeypatch.setattr(entry, "ensure_mcp_discovery_started", lambda: None) + monkeypatch.setattr(entry, "resolve_skin", lambda: "default") + monkeypatch.setattr(entry.server, "_ensure_skin_watcher", lambda: None) + monkeypatch.setattr(entry, "_log_exit", lambda reason: None) + monkeypatch.setattr(entry, "handle_spurious_eof", lambda *a: False) + monkeypatch.setattr(entry, "write_json", lambda _payload: True) + monkeypatch.setattr(entry.sys, "stdin", io.StringIO("")) + + # Prewarm is imported lazily inside main(); keep it inert. + import hermes_cli.model_switch as ms + + monkeypatch.setattr(ms, "prewarm_picker_cache_async", lambda: None) + + entry.main() + assert scheduled["n"] == 1 + + def test_handle_ws_schedules_sweep(self, monkeypatch): + import asyncio + + from tui_gateway import ws as ws_mod + + scheduled = {"n": 0} + monkeypatch.setattr( + server, "_schedule_startup_orphan_sweep", lambda: scheduled.__setitem__("n", scheduled["n"] + 1) + ) + monkeypatch.setattr(server, "resolve_skin", lambda: "default") + monkeypatch.setattr(server, "_ensure_skin_watcher", lambda: None) + monkeypatch.setattr(server, "register_live_transport", lambda *_a, **_k: None) + monkeypatch.setattr(server, "_WS_ORPHAN_REAP_GRACE_S", 0) + + class FakeWS: + async def accept(self): + pass + + async def send_text(self, line): + pass + + async def receive_text(self): + raise ws_mod._WebSocketDisconnect() + + async def close(self): + pass + + asyncio.run(ws_mod.handle_ws(FakeWS())) + assert scheduled["n"] == 1 + + def test_schedule_failure_does_not_break_main(self, monkeypatch): + def _boom(): + raise RuntimeError("nope") + + monkeypatch.setattr(server, "_schedule_startup_orphan_sweep", _boom) + monkeypatch.setattr(entry, "_install_sidecar_publisher", lambda: None) + monkeypatch.setattr(entry, "ensure_mcp_discovery_started", lambda: None) + monkeypatch.setattr(entry, "resolve_skin", lambda: "default") + monkeypatch.setattr(entry.server, "_ensure_skin_watcher", lambda: None) + monkeypatch.setattr(entry, "_log_exit", lambda reason: None) + monkeypatch.setattr(entry, "handle_spurious_eof", lambda *a: False) + monkeypatch.setattr(entry, "write_json", lambda _payload: True) + monkeypatch.setattr(entry.sys, "stdin", io.StringIO("")) + import hermes_cli.model_switch as ms + + monkeypatch.setattr(ms, "prewarm_picker_cache_async", lambda: None) + + entry.main() # must not raise diff --git a/tui_gateway/entry.py b/tui_gateway/entry.py index 00b801011b306..a7e7b3d48c00e 100644 --- a/tui_gateway/entry.py +++ b/tui_gateway/entry.py @@ -421,6 +421,16 @@ def ensure_mcp_discovery_started() -> None: def main(): _install_sidecar_publisher() + # One-time sweep of session rows orphaned by a previous gateway process + # (#65194) — the in-process WS-orphan reap timer dies with the process. + # Desktop/dashboard reach the agent through handle_ws instead; the + # scheduler is once-per-process + config-gated so the second site is a + # no-op when this already ran. + try: + server._schedule_startup_orphan_sweep() + except Exception: + logger.warning("startup orphan sweep scheduling failed", exc_info=True) + # MCP tool discovery — backgrounded so a slow or unreachable MCP server # can't freeze TUI startup (a dead stdio/http server burns 1+2+4s of # connect retries → ~7s of dead air before the composer appears). The diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 9493814bd0792..1258222a96a85 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -1393,6 +1393,116 @@ def _loop(): threading.Thread(target=_loop, daemon=True).start() +# ── Startup sweep for orphaned session rows (#65194) ───────────────────── +# The WS-orphan reaper above is an in-process threading.Timer: a gateway +# restart (update, crash, systemd) kills it before it fires, leaving the +# session row `ended_at IS NULL` forever. This is the startup complement +# every other resource type already has (docker_orphan_reaper, compression +# orphans). Scheduled once per process from both gateway entry points +# (stdio `entry.main` and the WS sidecar's `handle_ws`) — desktop/dashboard +# never run `entry.main()`. state.db is shared by sibling processes on the +# same profile, so eligibility is conservative. Disable via +# `sessions.orphan_reaper: false` (default on). +_ORPHAN_SWEEP_SOURCES = ("tui", "desktop", "subagent") +_startup_orphan_sweep_ran = False +_startup_orphan_sweep_lock = threading.Lock() + + +def _session_orphan_reaper_enabled() -> bool: + """``sessions.orphan_reaper`` (default on). Fail-open on config errors.""" + try: + sessions_cfg = (_load_cfg() or {}).get("sessions") or {} + if isinstance(sessions_cfg, dict) and "orphan_reaper" in sessions_cfg: + return is_truthy_value(sessions_cfg.get("orphan_reaper"), default=True) + # Fail-open: a missing key (raw yaml, no DEFAULT_CONFIG merge on + # this loader) must keep the sweep on. + return True + except Exception: + return True + + +def _live_session_ids() -> list[str]: + """Session ids this process currently holds in memory.""" + ids: set[str] = set() + with _sessions_lock: + for sid, session in _sessions.items(): + if sid: + ids.add(str(sid)) + agent = session.get("agent") if isinstance(session, dict) else None + for candidate in ( + getattr(agent, "session_id", None), + session.get("session_key") if isinstance(session, dict) else None, + ): + if candidate: + ids.add(str(candidate)) + return sorted(ids) + + +def _sweep_orphaned_session_rows() -> list[str]: + """End orphaned tui/desktop/subagent rows left by a dead process. + + "Provably orphaned" is inferred conservatively from inactivity — the + row must have been created AND last messaged at least the session TTL + ago (``HERMES_TUI_SESSION_TTL_S``). A freshly created row that copied + an old transcript is protected by its own ``started_at``. Rows this + process still holds in memory (e.g. a ``session.resume`` during the + startup grace window) are excluded so the sweep never races a + mid-reconnect client. + """ + db = _get_db() + if db is None: + return [] + ttl = _SESSION_TTL_S + if ttl <= 0: + return [] + swept = db.sweep_orphaned_sessions( + max_idle_seconds=ttl, + sources=_ORPHAN_SWEEP_SOURCES, + exclude_ids=tuple(_live_session_ids()), + ) + if swept: + logger.info( + "Closed %d orphaned session row(s) from a previous gateway " + "process (startup_orphan_reap): %s", + len(swept), + ", ".join(swept), + ) + return swept + + +def _schedule_startup_orphan_sweep() -> None: + """Schedule the once-per-process startup orphan sweep (#65194). + + Called from both gateway entry points. Repeat calls are no-ops. The + sweep is delayed by the WS-orphan grace window so a client reconnecting + right after a restart can ``session.resume`` its row before the sweep + reads the DB. ``HERMES_TUI_WS_ORPHAN_REAP_GRACE_S=0`` (park forever) + and ``HERMES_TUI_SESSION_TTL_S=0`` both suppress the sweep; so does + ``sessions.orphan_reaper: false``. + """ + global _startup_orphan_sweep_ran + if _WS_ORPHAN_REAP_GRACE_S <= 0 or _SESSION_TTL_S <= 0: + return + if not _session_orphan_reaper_enabled(): + return + if _startup_orphan_sweep_ran: + return + with _startup_orphan_sweep_lock: + if _startup_orphan_sweep_ran: + return + _startup_orphan_sweep_ran = True + + def _run() -> None: + try: + _sweep_orphaned_session_rows() + except Exception: + logger.warning("startup orphan session sweep failed", exc_info=True) + + timer = threading.Timer(_WS_ORPHAN_REAP_GRACE_S, _run) + timer.daemon = True + timer.start() + + atexit.register(_shutdown_sessions) _start_idle_reaper() diff --git a/tui_gateway/ws.py b/tui_gateway/ws.py index 073c4ac1497c7..2afdd994d68c3 100644 --- a/tui_gateway/ws.py +++ b/tui_gateway/ws.py @@ -330,6 +330,15 @@ async def handle_ws(ws: Any) -> None: # Track this peer for session-less global broadcasts (skin.changed # from the background watcher) — write_json can't route those. server.register_live_transport(transport) + # Same once-per-process startup pass for session rows orphaned by a + # previous gateway process (#65194): the desktop app and web dashboard + # reach the agent through this WS sidecar, not entry.main(). Idempotent + # + config-gated inside, so a stdio TUI that already scheduled is a + # no-op. + try: + server._schedule_startup_orphan_sweep() + except Exception: + _log.warning("startup orphan sweep scheduling failed", exc_info=True) if not ready_ok: disconnect_reason = "ready_send_failed" send_failures += 1 diff --git a/website/docs/user-guide/configuration.md b/website/docs/user-guide/configuration.md index 0eccf2d5aa519..4786005a00fc4 100644 --- a/website/docs/user-guide/configuration.md +++ b/website/docs/user-guide/configuration.md @@ -2108,6 +2108,17 @@ group_sessions_per_user: true # true = per-user isolation in groups/channels, f For the behavior details and examples, see [Sessions](/user-guide/sessions) and the [Discord guide](/user-guide/messaging/discord). +## Session Maintenance + +Session rows live in `~/.hermes/state.db`. Automatic maintenance is configured under `sessions` in `config.yaml`: + +```yaml +sessions: + orphan_reaper: true # Close tui/desktop/subagent rows orphaned by a dead gateway process +``` + +**Orphan reaper** (`orphan_reaper`, default `true`): the TUI/desktop gateway normally closes a disconnected session's row after a short in-process grace timer. If the gateway restarts (update, crash, systemd) before the timer fires, the row stays open forever and shows up as phantom "active" work in `/resume` and dashboards. On every gateway boot — both the stdio TUI (`entry.main`) and the desktop/dashboard WebSocket sidecar (`handle_ws`) — rows with source `tui` / `desktop` / `subagent` whose start time **and** newest message are both older than the session TTL (`HERMES_TUI_SESSION_TTL_S`, default 6 hours) are closed with `end_reason: startup_orphan_reap`. Messaging-platform sessions (Telegram, Discord, …) are never touched, live in-memory sessions (a client that already resumed) are excluded, and swept sessions remain resumable. + ## Unauthorized DM Behavior Control what Hermes does when an unknown user sends a direct message: From fad9997e2bf77fb7172efc7b054f34801f7a369c Mon Sep 17 00:00:00 2001 From: halaprix <6533433+halaprix@users.noreply.github.com> Date: Wed, 19 Aug 2026 09:44:36 +0200 Subject: [PATCH 2/2] chore: refresh PR head after rebase onto current main