171 lines
7.2 KiB
Python
171 lines
7.2 KiB
Python
"""Idle-exit for Desktop-owned ``hermes serve --isolated`` backends reached over SSH (#101626).
|
|
|
|
That backend is deliberately detached (``setsid``/``nohup``, PPID 1) so it survives the SSH channel
|
|
closing, and every teardown path lives on the CLIENT. A laptop that sleeps mid-session (dark wake
|
|
reconnects the tunnel, spawns a backend, sleeps again) therefore leaves a backend behind every
|
|
cycle — each one an extra writer on ``state.db``. The server needs its own liveness signal.
|
|
|
|
Two pieces, both scoped to the SSH-isolated case (a session token was handed over via
|
|
``--ssh-session-token-file``):
|
|
|
|
* An ASGI wrapper counts accepted WebSocket connections (every dashboard WS route: /api/ws,
|
|
/api/pty, /api/console, /api/pub, /api/events, /api/audio/speak-stream) without touching the
|
|
handlers. When the count has been zero for the grace window and no agent turn is running, the
|
|
watchdog asks uvicorn to exit gracefully (WAL checkpoint, exit 0). An indeterminate turn probe
|
|
fails closed: the backend stays up.
|
|
* Loopback normally disables uvicorn's WS ping (a dead local client sends FIN/RST). Across an SSH
|
|
tunnel the local socket is healthy while the far end is asleep, so pings are the only way to notice
|
|
a half-open tunnel; the isolated backend keeps a slow ping with a long timeout so a GIL-holding
|
|
turn cannot trip it.
|
|
|
|
Design and the client-count/turn-probe/fail-closed shape are from #101678 by @StanleyStetson; this
|
|
is the slim redo on the decomposed web server.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import threading
|
|
import time
|
|
from typing import Callable, Optional
|
|
|
|
_log = logging.getLogger(__name__)
|
|
|
|
DEFAULT_IDLE_GRACE_S = 900.0
|
|
# Slow enough that a long GIL-holding turn (minutes) cannot trip it, fast enough that a sleeping
|
|
# laptop's half-open tunnel is noticed well inside the idle grace window.
|
|
TUNNEL_WS_PING_INTERVAL_S = 60.0
|
|
TUNNEL_WS_PING_TIMEOUT_S = 600.0
|
|
|
|
|
|
class IdleClientTracker:
|
|
"""Live accepted-WebSocket count plus the moment the last client left."""
|
|
|
|
def __init__(self, now: Callable[[], float] = time.monotonic) -> None:
|
|
self._now = now
|
|
self._lock = threading.Lock()
|
|
self._live = 0
|
|
self._last_client_at = now()
|
|
|
|
def on_open(self) -> None:
|
|
with self._lock:
|
|
self._live += 1
|
|
self._last_client_at = self._now()
|
|
|
|
def on_close(self) -> None:
|
|
with self._lock:
|
|
self._live = max(0, self._live - 1)
|
|
self._last_client_at = self._now()
|
|
|
|
def live_count(self) -> int:
|
|
with self._lock:
|
|
return self._live
|
|
|
|
def idle_for(self) -> float:
|
|
with self._lock:
|
|
return 0.0 if self._live else self._now() - self._last_client_at
|
|
|
|
|
|
def wrap_asgi_with_ws_tracking(app, tracker: IdleClientTracker):
|
|
"""Count WebSocket sessions at the ASGI boundary: open on the ``websocket.accept`` send, close
|
|
when the scope ends. Handlers stay untouched, so a new WS route is tracked automatically."""
|
|
|
|
async def _app(scope, receive, send):
|
|
if scope.get("type") != "websocket":
|
|
return await app(scope, receive, send)
|
|
accepted = False
|
|
|
|
async def _send(message):
|
|
nonlocal accepted
|
|
if message.get("type") == "websocket.accept" and not accepted:
|
|
accepted = True
|
|
tracker.on_open()
|
|
await send(message)
|
|
|
|
try:
|
|
await app(scope, receive, _send)
|
|
finally:
|
|
if accepted:
|
|
tracker.on_close()
|
|
|
|
return _app
|
|
|
|
|
|
_probe_failure_logged = False
|
|
|
|
|
|
def _session_work_in_flight(session: dict) -> bool:
|
|
if any(session.get(key) for key in ("running", "queued_prompt", "queued_prompts", "_auto_continue_scheduled")):
|
|
return True
|
|
return any(thread.is_alive() for key in ("_run_thread", "_agent_build_thread")
|
|
if (thread := session.get(key)) is not None)
|
|
|
|
|
|
def busy_ledger() -> Optional[str]:
|
|
"""Name the ledger that holds work — ``"retirement_admission"``, ``"session:<id>"``,
|
|
``"delegation"``, ``"cron:<job ids>"`` — ``""`` when every ledger is empty, ``None`` when one
|
|
cannot be read. The Desktop's idle probe reports this so a backend that will not retire says
|
|
WHICH work it is protecting, instead of an opaque "turn in flight".
|
|
|
|
Session work lives on ``tui_gateway.server`` (the voice mixin's helper is bound into that
|
|
namespace). Cron runs live outside that table (``cron.scheduler.get_running_job_ids``, the same
|
|
ledger the gateway shutdown drain reads): without it a daily job mid-run reported "no turn" and
|
|
the exit killed it (#107485). None keeps the backend alive forever, so the cause is logged once —
|
|
a silent never-exits would be the original bug with a new face."""
|
|
global _probe_failure_logged
|
|
try:
|
|
import tui_gateway.server as gateway
|
|
from hermes_cli.backend_retirement import retirement
|
|
|
|
if retirement.active_count():
|
|
return "retirement_admission"
|
|
with gateway._sessions_lock:
|
|
busy_sessions = [sid for sid, s in gateway._sessions.items() if _session_work_in_flight(s)]
|
|
if busy_sessions:
|
|
return "session:" + ",".join(str(sid) for sid in busy_sessions)
|
|
from tools.async_delegation import active_count
|
|
from cron.scheduler import get_running_job_ids
|
|
if active_count():
|
|
return "delegation"
|
|
running_jobs = get_running_job_ids()
|
|
return "cron:" + ",".join(sorted(running_jobs)) if running_jobs else ""
|
|
except Exception:
|
|
if not _probe_failure_logged:
|
|
_probe_failure_logged = True
|
|
_log.warning("idle-exit turn probe unavailable; this backend will not self-retire", exc_info=True)
|
|
return None
|
|
|
|
|
|
def turn_in_flight() -> Optional[bool]:
|
|
"""Bool view of :func:`busy_ledger`, kept so ``should_exit_idle`` / the idle watchdog and injected
|
|
test probes keep their ``Optional[bool]`` contract; None when the ledgers cannot be read."""
|
|
ledger = busy_ledger()
|
|
return None if ledger is None else bool(ledger)
|
|
|
|
|
|
def should_exit_idle(tracker: IdleClientTracker, grace_s: float,
|
|
probe: Callable[[], Optional[bool]] = turn_in_flight) -> bool:
|
|
"""Exit only when no client has been connected for ``grace_s`` AND no turn is provably running.
|
|
A probe that cannot answer keeps the process (fail closed)."""
|
|
return tracker.idle_for() >= grace_s and probe() is False # idle_for() is 0 while a client is connected
|
|
|
|
|
|
def start_idle_watchdog(server, tracker: IdleClientTracker, *, grace_s: float = DEFAULT_IDLE_GRACE_S,
|
|
poll_s: float = 15.0, probe: Callable[[], Optional[bool]] = turn_in_flight) -> threading.Thread:
|
|
"""Daemon thread that sets ``server.should_exit`` once :func:`should_exit_idle` holds."""
|
|
|
|
poll_s = min(poll_s, max(0.5, grace_s / 4))
|
|
|
|
def _loop() -> None:
|
|
while not getattr(server, "should_exit", False):
|
|
if should_exit_idle(tracker, grace_s, probe):
|
|
_log.warning("SSH-isolated backend idle for %.0fs with no client and no running turn; exiting.",
|
|
tracker.idle_for())
|
|
server.should_exit = True
|
|
return
|
|
time.sleep(poll_s)
|
|
|
|
thread = threading.Thread(target=_loop, daemon=True, name="ssh-isolated-idle-watchdog")
|
|
thread.start()
|
|
return thread
|