285 lines
11 KiB
Python
285 lines
11 KiB
Python
"""Scale-to-zero: an attached dashboard/desktop/TUI WS client counts as activity.
|
|
|
|
Background (2026-09-02 fleet audit): 13 of 72 active opted-in prod instances
|
|
flapped suspend -> proxy-wake every ~60s. The gateway only stamped
|
|
``_last_inbound_at`` for messaging inbound, so it suspended under an open
|
|
dashboard client; the client's reconnect loop re-poked the Fly-proxied hostname
|
|
and autostart resumed the box. The dashboard runs in a separate process on
|
|
hosted instances, so the signal crosses over as a marker-file mtime.
|
|
|
|
These tests exercise the REAL seams — the pure helpers with a real temp
|
|
HERMES_HOME, GatewayRunner._scale_to_zero_is_idle's composition, and
|
|
tui_gateway.ws.handle_ws — rather than stubbing the collection under test
|
|
(the F25 / #84327 lesson: bugs live at the call site, not in the pure predicate).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from gateway import scale_to_zero as s2z
|
|
from gateway.run import GatewayRunner
|
|
|
|
|
|
@pytest.fixture
|
|
def hermes_home(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
|
return tmp_path
|
|
|
|
|
|
def _stat_denying(target):
|
|
"""os.stat replacement that denies ONLY the marker path (stdlib callers unaffected)."""
|
|
real = os.stat
|
|
|
|
def _stat(path, *a, **k):
|
|
if os.fspath(path) == os.fspath(target):
|
|
raise PermissionError("nope")
|
|
return real(path, *a, **k)
|
|
|
|
return _stat
|
|
|
|
|
|
# --- pure helpers -----------------------------------------------------------
|
|
|
|
|
|
def test_heartbeat_path_lives_under_hermes_home_state(hermes_home):
|
|
p = s2z.dashboard_client_heartbeat_path()
|
|
assert p == hermes_home / "state" / "dashboard_clients.heartbeat"
|
|
|
|
|
|
def test_last_seen_missing_marker_is_none_not_fail_awake(hermes_home):
|
|
# Steady state for a box nobody has the dashboard open on: must read as
|
|
# "no client", otherwise no instance would ever suspend.
|
|
assert s2z.dashboard_client_last_seen() is None
|
|
|
|
|
|
def test_touch_creates_state_dir_and_marker(hermes_home):
|
|
assert s2z.touch_dashboard_client_heartbeat() is True
|
|
p = s2z.dashboard_client_heartbeat_path()
|
|
assert p.exists()
|
|
seen = s2z.dashboard_client_last_seen()
|
|
assert seen is not None and abs(seen - time.time()) < 5
|
|
|
|
|
|
def test_last_seen_returns_raw_mtime_without_staleness_cutoff(hermes_home):
|
|
# No liveness cutoff here on purpose: is_idle decides recency. A 1h-old
|
|
# marker still reports its mtime; the gateway then finds it outside
|
|
# idle_timeout, same as an old _last_inbound_at.
|
|
s2z.touch_dashboard_client_heartbeat()
|
|
p = s2z.dashboard_client_heartbeat_path()
|
|
mtime = os.stat(p).st_mtime
|
|
assert s2z.dashboard_client_last_seen(now=mtime + 10) == mtime
|
|
assert s2z.dashboard_client_last_seen(now=mtime + 3600) == mtime
|
|
|
|
|
|
def test_last_seen_future_mtime_is_clamped_to_now(hermes_home):
|
|
# A wall-clock step-back can leave the marker in the future; it must not
|
|
# extend the idle window past "now".
|
|
s2z.touch_dashboard_client_heartbeat()
|
|
p = s2z.dashboard_client_heartbeat_path()
|
|
future = time.time() + 600
|
|
os.utime(p, (future, future))
|
|
now = time.time()
|
|
assert s2z.dashboard_client_last_seen(now=now) == now
|
|
|
|
|
|
def test_last_seen_unreadable_marker_fails_awake(hermes_home, monkeypatch):
|
|
s2z.touch_dashboard_client_heartbeat()
|
|
|
|
monkeypatch.setattr(s2z.os, "stat", _stat_denying(s2z.dashboard_client_heartbeat_path()))
|
|
now = 1_000_000.0
|
|
# Unreadable (not missing) => counts as activity right now.
|
|
assert s2z.dashboard_client_last_seen(now=now) == now
|
|
|
|
|
|
def test_touch_never_raises(hermes_home, monkeypatch):
|
|
monkeypatch.setattr(s2z.os, "utime", lambda *a, **k: (_ for _ in ()).throw(OSError("ro")))
|
|
assert s2z.touch_dashboard_client_heartbeat() is False
|
|
|
|
|
|
# --- gateway side: the idle predicate composition ---------------------------
|
|
|
|
|
|
def _runner(monkeypatch, *, last_inbound_at):
|
|
r = GatewayRunner.__new__(GatewayRunner)
|
|
r._running = True
|
|
r._last_inbound_at = last_inbound_at
|
|
r._running_agents = {}
|
|
r._background_tasks = set()
|
|
r.adapters = {}
|
|
monkeypatch.setattr(r, "_scale_to_zero_idle_timeout_seconds", lambda: 120.0, raising=False)
|
|
monkeypatch.setattr(r, "_scale_to_zero_has_live_background_work", lambda: False, raising=False)
|
|
monkeypatch.setattr("cron.scheduler.get_running_job_ids", lambda: [])
|
|
return r
|
|
|
|
|
|
def test_idle_without_dashboard_client_unchanged(hermes_home, monkeypatch):
|
|
r = _runner(monkeypatch, last_inbound_at=time.time() - 600)
|
|
assert r._scale_to_zero_is_idle() is True
|
|
|
|
|
|
def test_attached_dashboard_client_blocks_idle(hermes_home, monkeypatch):
|
|
r = _runner(monkeypatch, last_inbound_at=time.time() - 600)
|
|
s2z.touch_dashboard_client_heartbeat()
|
|
assert r._scale_to_zero_is_idle() is False
|
|
|
|
|
|
def test_client_gets_the_same_idle_grace_as_a_message(hermes_home, monkeypatch):
|
|
"""Last WS frame 100s ago with a 120s idle_timeout => still inside the
|
|
window => NOT idle. This is the 2-minute-after-the-app-closes contract; an
|
|
earlier draft cut the marker off at 45s and suspended ~50s after
|
|
disconnect (observed live on staging)."""
|
|
r = _runner(monkeypatch, last_inbound_at=time.time() - 600)
|
|
s2z.touch_dashboard_client_heartbeat()
|
|
p = s2z.dashboard_client_heartbeat_path()
|
|
old = time.time() - 100
|
|
os.utime(p, (old, old))
|
|
assert r._scale_to_zero_is_idle() is False
|
|
|
|
|
|
def test_client_gone_longer_than_idle_timeout_is_idle(hermes_home, monkeypatch):
|
|
r = _runner(monkeypatch, last_inbound_at=time.time() - 600)
|
|
s2z.touch_dashboard_client_heartbeat()
|
|
p = s2z.dashboard_client_heartbeat_path()
|
|
old = time.time() - 121
|
|
os.utime(p, (old, old))
|
|
assert r._scale_to_zero_is_idle() is True
|
|
|
|
|
|
def test_marker_predating_gateway_inbound_does_not_matter(hermes_home, monkeypatch):
|
|
# Ancient marker from a client that left hours ago, gateway idle 600s.
|
|
r = _runner(monkeypatch, last_inbound_at=time.time() - 600)
|
|
s2z.touch_dashboard_client_heartbeat()
|
|
p = s2z.dashboard_client_heartbeat_path()
|
|
old = time.time() - 7200
|
|
os.utime(p, (old, old))
|
|
assert r._scale_to_zero_is_idle() is True
|
|
|
|
|
|
def test_dashboard_client_seen_recently_extends_inbound_clock(hermes_home, monkeypatch):
|
|
# Marker 30s old: inbound clock moves to 30s ago, which is
|
|
# inside the 120s window => not idle, even though the gateway's own
|
|
# _last_inbound_at is ancient.
|
|
r = _runner(monkeypatch, last_inbound_at=time.time() - 600)
|
|
s2z.touch_dashboard_client_heartbeat()
|
|
p = s2z.dashboard_client_heartbeat_path()
|
|
t = time.time() - 30
|
|
os.utime(p, (t, t))
|
|
assert r._scale_to_zero_is_idle() is False
|
|
|
|
|
|
def test_newer_gateway_inbound_wins_over_older_marker(hermes_home, monkeypatch):
|
|
r = _runner(monkeypatch, last_inbound_at=time.time() - 5)
|
|
monkeypatch.setattr(r, "_scale_to_zero_idle_timeout_seconds", lambda: 10.0, raising=False)
|
|
s2z.touch_dashboard_client_heartbeat()
|
|
p = s2z.dashboard_client_heartbeat_path()
|
|
t = time.time() - 40
|
|
os.utime(p, (t, t))
|
|
# Picking the marker (40s > 10s) would read idle; the chat message (5s) wins.
|
|
assert r._scale_to_zero_is_idle() is False
|
|
# _last_inbound_at itself is not mutated by the read.
|
|
assert time.time() - r._last_inbound_at < 10
|
|
|
|
|
|
def test_unreadable_marker_keeps_gateway_awake(hermes_home, monkeypatch):
|
|
r = _runner(monkeypatch, last_inbound_at=time.time() - 600)
|
|
s2z.touch_dashboard_client_heartbeat()
|
|
monkeypatch.setattr(s2z.os, "stat", _stat_denying(s2z.dashboard_client_heartbeat_path()))
|
|
assert r._scale_to_zero_is_idle() is False
|
|
|
|
|
|
# --- dashboard side: the real handle_ws path touches the marker -------------
|
|
|
|
|
|
def test_handle_ws_connect_touches_marker(hermes_home, monkeypatch):
|
|
from tui_gateway import server, ws as ws_mod
|
|
|
|
monkeypatch.setattr(server, "_start_backend_heartbeat_refresher", lambda: None)
|
|
monkeypatch.setattr(server, "_schedule_startup_orphan_sweep", lambda: None, raising=False)
|
|
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)
|
|
monkeypatch.setattr(ws_mod, "_dashboard_client_touched_at", 0.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
|
|
|
|
assert s2z.dashboard_client_last_seen() is None
|
|
asyncio.run(ws_mod.handle_ws(FakeWS()))
|
|
seen = s2z.dashboard_client_last_seen()
|
|
assert seen is not None and abs(seen - time.time()) < 5
|
|
|
|
|
|
def test_handle_ws_inbound_frames_refresh_marker(hermes_home, monkeypatch):
|
|
from tui_gateway import server, ws as ws_mod
|
|
|
|
monkeypatch.setattr(server, "_start_backend_heartbeat_refresher", lambda: None)
|
|
monkeypatch.setattr(server, "_schedule_startup_orphan_sweep", lambda: None, raising=False)
|
|
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)
|
|
monkeypatch.setattr(ws_mod, "_dashboard_client_touched_at", 0.0)
|
|
# Disable the throttle so each frame is observable.
|
|
monkeypatch.setattr(ws_mod, "_DASHBOARD_CLIENT_TOUCH_MIN_INTERVAL_S", 0.0)
|
|
|
|
frames = ['{"jsonrpc":"2.0","method":"gateway.ping","id":1}'] * 2
|
|
touches: list[float] = []
|
|
real_touch = s2z.touch_dashboard_client_heartbeat
|
|
|
|
def _spy(path=None):
|
|
touches.append(time.time())
|
|
return real_touch(path)
|
|
|
|
monkeypatch.setattr(s2z, "touch_dashboard_client_heartbeat", _spy)
|
|
|
|
class FakeWS:
|
|
async def accept(self):
|
|
pass
|
|
|
|
async def send_text(self, line):
|
|
pass
|
|
|
|
async def receive_text(self):
|
|
if frames:
|
|
return frames.pop()
|
|
raise ws_mod._WebSocketDisconnect()
|
|
|
|
async def close(self):
|
|
pass
|
|
|
|
asyncio.run(ws_mod.handle_ws(FakeWS()))
|
|
# 1 on connect + 1 per inbound frame.
|
|
assert len(touches) == 3
|
|
assert s2z.dashboard_client_last_seen() is not None
|
|
|
|
|
|
def test_note_activity_is_throttled(hermes_home, monkeypatch):
|
|
from tui_gateway import ws as ws_mod
|
|
|
|
calls = {"n": 0}
|
|
monkeypatch.setattr(
|
|
s2z, "touch_dashboard_client_heartbeat", lambda path=None: calls.__setitem__("n", calls["n"] + 1) or True
|
|
)
|
|
monkeypatch.setattr(ws_mod, "_dashboard_client_touched_at", 0.0)
|
|
ws_mod._note_dashboard_client_activity(force=True)
|
|
ws_mod._note_dashboard_client_activity()
|
|
ws_mod._note_dashboard_client_activity()
|
|
assert calls["n"] == 1
|
|
ws_mod._note_dashboard_client_activity(force=True)
|
|
assert calls["n"] == 2
|