"""Unit tests for the scale-to-zero idle-detection pure logic (Phase 0). Behaviour-contract tests (AGENTS.md): each conjunct of the idle predicate and each clause of the arm-gate is exercised independently, not frozen against a snapshot. The pure helpers in gateway/scale_to_zero.py take plain inputs so they test without a live gateway. """ from __future__ import annotations import pytest from gateway.scale_to_zero import ( DEFAULT_IDLE_TIMEOUT_MINUTES, SCALE_TO_ZERO_ENV, is_idle, messaging_is_relay_only_or_absent, parse_idle_timeout_seconds, scale_to_zero_enabled, should_arm, ) # ── scale_to_zero_enabled (the Labs HERMES_SCALE_TO_ZERO stamp, D11/Q8=A) ──── @pytest.mark.parametrize("value", ["1", "true", "TRUE", "yes", "on", " On "]) def test_enabled_truthy_values(value): assert scale_to_zero_enabled({SCALE_TO_ZERO_ENV: value}) is True # ── parse_idle_timeout_seconds (config.yaml, D2) ───────────────────────────── def test_timeout_parses_minutes_to_seconds(): assert parse_idle_timeout_seconds(5) == 300.0 assert parse_idle_timeout_seconds(10) == 600.0 assert parse_idle_timeout_seconds("5") == 300.0 def test_timeout_invalid_values_degrade_to_default(): # Behavior contract: bad config falls back to the module default (whatever # its current value), never to zero/negative — an instant-dormant gateway # is never the intent. for bad in (None, "", "nope", 0, -3): assert parse_idle_timeout_seconds(bad) == DEFAULT_IDLE_TIMEOUT_MINUTES * 60.0 # ── messaging_is_relay_only_or_absent (F6/D1) ──────────────────────────────── class _P: """Stand-in for a Platform enum member with a ``.value``.""" def __init__(self, value): self.value = value def test_relay_only_is_true(): assert messaging_is_relay_only_or_absent([_P("relay")]) is True def test_no_platform_is_true(): # A Chronos-only / no-messaging-platform agent can scale to zero. assert messaging_is_relay_only_or_absent([]) is True def test_scale_to_zero_gate_accounts_for_secondary_profile_direct_adapter(monkeypatch): # A direct adapter in a served profile owns an inbound socket too; the process # must stay awake even when the launch profile itself is relay-only. from gateway.config import GatewayConfig, Platform, PlatformConfig from gateway.run_shutdown import GatewayShutdownMixin monkeypatch.setenv("HERMES_SCALE_TO_ZERO", "1") monkeypatch.setenv("GATEWAY_RELAY_WAKE_URL", "https://wake.example.test/instance") runner = object.__new__(GatewayShutdownMixin) runner.config = GatewayConfig(platforms={Platform.RELAY: PlatformConfig(enabled=True)}) runner.adapters = {Platform.RELAY: object()} runner._profile_adapters = {"imessage": {Platform("photon"): object()}} assert runner._scale_to_zero_should_arm() is False # Mid-reconnect the adapter is popped from _profile_adapters and parked in # _profile_failed_platforms (retryable fatal); it is still served and must # keep the process awake or the reconnect can never complete. runner._profile_adapters = {} runner._profile_failed_platforms = {"imessage": {Platform("photon"): object()}} assert runner._scale_to_zero_should_arm() is False def test_watcher_reasks_gate_before_dormant_sequence(monkeypatch): # Arming is a boot-time snapshot. A direct adapter hot-added afterwards (profile reconcile) # must stop the dormant sequence; with relay only, the same tick still reaches go_dormant. import asyncio import time from gateway.config import GatewayConfig, Platform, PlatformConfig from gateway.run_shutdown import GatewayShutdownMixin monkeypatch.setenv("GATEWAY_RELAY_SLEEP_URL", "https://sleep.example.test/instance") class Relay: def __init__(self): self.calls = [] def hold_redial(self): return True def release_redial(self): return True async def go_dormant(self): self.calls.append("go_dormant") return False # refuse the ack: the tick must abandon before any suspend call class Runner(GatewayShutdownMixin): def __init__(self): self.config = GatewayConfig(platforms={Platform.RELAY: PlatformConfig(enabled=True)}) self.adapters = {Platform.RELAY: Relay()} self._profile_adapters = {} self._running = True self._running_agents = {} self._background_tasks = set() self._last_inbound_at = time.time() - 3600 self._scale_to_zero_cooldown_until = 0.0 self._scale_to_zero_no_suspend_logged = False self._scale_to_zero_direct_platform_logged = False def _running_agent_count(self): return 0 def _update_runtime_status(self, state): pass async def one_tick(runner): task = asyncio.ensure_future(runner._scale_to_zero_watcher(interval=0.0)) await asyncio.sleep(0.05) runner._running = False task.cancel() try: await task except asyncio.CancelledError: pass hot_added = Runner() hot_added._profile_adapters = {"imessage": {Platform("photon"): object()}} asyncio.run(one_tick(hot_added)) assert hot_added.adapters[Platform.RELAY].calls == [] relay_only = Runner() asyncio.run(one_tick(relay_only)) assert relay_only.adapters[Platform.RELAY].calls == ["go_dormant"] # ── should_arm (D1/D11/§3.4(1)) ────────────────────────────────────────────── def test_arm_blocked_without_wake_url(): # A suspended instance with no wake target is a black hole (§3.4(1)). assert should_arm(enabled=True, relay_only_or_absent=True, wake_url=None) is False assert should_arm(enabled=True, relay_only_or_absent=True, wake_url="") is False # ── is_idle (D2/D3/F7) — each conjunct flips the result ────────────────────── def _idle_kwargs(**over): base = dict( active_work_count=0, seconds_since_last_inbound=600.0, idle_timeout_seconds=300.0, has_live_background_work=False, ) base.update(over) return base def test_not_idle_with_running_agent(): assert is_idle(**_idle_kwargs(active_work_count=1)) is False def test_idle_exactly_at_threshold(): # >= timeout is idle (boundary). assert is_idle(**_idle_kwargs(seconds_since_last_inbound=300.0)) is True # ── suspend_self / self_suspend_available (the gateway-owned suspend call) ─── # # Fly Proxy autostop is inbound-only and job-blind (and since mid-2026 no longer # counts outbound sockets as activity), so the gateway suspends its own machine # via the local flaps unix socket strictly after the idle predicate + dormant # quiesce. These exercise the wire call against a real unix-socket fake flaps. import os import shutil import socket as _socket import tempfile import threading from pathlib import Path from gateway.scale_to_zero import ( # noqa: E402 - grouped with their section FLY_APP_NAME_ENV, FLY_MACHINE_ID_ENV, self_suspend_available, suspend_self, ) _FLY_ENV = {FLY_APP_NAME_ENV: "hermes-agent-stg-test", FLY_MACHINE_ID_ENV: "d891234f"} # sockaddr_un.sun_path is 104 bytes on macOS/BSD and 108 on Linux (incl. NUL). _SUN_PATH_MAX = 100 @pytest.fixture() def short_sock_dir(): """A directory short enough that ``/fly-api.sock`` fits in sun_path. pytest's ``tmp_path`` nests deep enough on macOS and on CI runners that a socket bound under it fails with ``OSError: AF_UNIX path too long``. The production path is the fixed ``/.fly/api`` so the length limit never applies there; only the fake flaps server needs a short home. """ candidates = [tempfile.gettempdir(), "/tmp"] for base in candidates: try: path = Path(tempfile.mkdtemp(prefix="s2z-", dir=base)) except OSError: continue if len(str(path / "fly-api.sock").encode()) <= _SUN_PATH_MAX: break shutil.rmtree(path, ignore_errors=True) else: pytest.skip(f"no temp dir short enough for an AF_UNIX socket (tried {candidates})") try: yield path finally: shutil.rmtree(path, ignore_errors=True) def _fake_flaps(sock_dir, status_line, capture): """One-shot unix-socket HTTP server standing in for flaps.""" sock_path = str(sock_dir / "fly-api.sock") server = _socket.socket(_socket.AF_UNIX, _socket.SOCK_STREAM) server.bind(sock_path) server.listen(1) def serve(): conn, _ = server.accept() with conn: conn.settimeout(5) data = b"" while b"\r\n\r\n" not in data: chunk = conn.recv(4096) if not chunk: break data += chunk capture.append(data) conn.sendall( f"HTTP/1.1 {status_line}\r\nContent-Length: 2\r\nConnection: close\r\n\r\n{{}}".encode() ) server.close() t = threading.Thread(target=serve, daemon=True) t.start() return sock_path, t @pytest.mark.platforms("linux") def test_suspend_self_posts_suspend_for_this_machine(short_sock_dir): captured: list[bytes] = [] sock_path, t = _fake_flaps(short_sock_dir, "200 OK", captured) assert suspend_self(_FLY_ENV, socket_path=sock_path) is True t.join(timeout=5) request = captured[0].decode() # The request must target THIS machine's suspend endpoint, per the Fly # Machines API (POST /v1/apps/{app}/machines/{id}/suspend on /.fly/api). assert request.startswith( "POST /v1/apps/hermes-agent-stg-test/machines/d891234f/suspend HTTP/1.1\r\n" ) assert "Host: flaps\r\n" in request @pytest.mark.platforms("linux") def test_suspend_self_non_2xx_is_false_not_raise(short_sock_dir): captured: list[bytes] = [] sock_path, t = _fake_flaps(short_sock_dir, "412 Precondition Failed", captured) assert suspend_self(_FLY_ENV, socket_path=sock_path) is False t.join(timeout=5) @pytest.mark.platforms("linux") def test_suspend_self_missing_socket_is_false_not_raise(tmp_path): # Fail-awake: a dead/absent flaps socket must never raise out of the watcher. assert suspend_self(_FLY_ENV, socket_path=str(tmp_path / "nope.sock")) is False def test_suspend_self_requires_machine_identity(tmp_path): assert suspend_self({}, socket_path=str(tmp_path / "unused.sock")) is False def test_self_suspend_available_needs_identity_and_socket(): # No socket at /.fly/api in a test environment -> unavailable even with env. if not os.path.exists("/.fly/api"): assert self_suspend_available(_FLY_ENV) is False # Missing identity -> unavailable regardless of socket. assert self_suspend_available({}) is False # Brokered suspend: Azure's stop verb needs a credential the sandbox lacks, so # NAS stamps a signed sleep URL and stops the machine on our POST. from gateway.scale_to_zero import ( # noqa: E402 - grouped with their section SLEEP_URL_ENV, brokered_sleep_url, request_brokered_suspend, suspend_available, ) _SLEEP_URL = "https://portal.example.com/api/agents/inst-1/sleep?t=sig" def test_brokered_sleep_url_reads_the_stamp(): assert brokered_sleep_url({SLEEP_URL_ENV: _SLEEP_URL}) == _SLEEP_URL assert brokered_sleep_url({}) is None assert brokered_sleep_url({SLEEP_URL_ENV: " "}) is None def test_malformed_sleep_url_warns_once_not_every_idle_tick(monkeypatch, caplog): """The watcher calls this on every tick and its own no-lever latch sits after it, so an unlatched warning would repeat for the life of the process.""" import gateway.scale_to_zero as sz monkeypatch.setattr(sz, "_malformed_sleep_url_logged", False) monkeypatch.setenv("GATEWAY_RELAY_SLEEP_URL", "http://portal.example.com/x") with caplog.at_level("WARNING"): for _ in range(5): assert sz.brokered_sleep_url() is None assert sum("ignoring malformed" in r.message for r in caplog.records) == 1 def test_brokered_sleep_url_rejects_a_malformed_stamp(monkeypatch): """Validated before it is treated as a lever: otherwise the watcher marks draining, holds the re-dial and flips the connector, and only then does urllib reject the value, quiescing for a suspend that could never happen.""" from gateway.scale_to_zero import brokered_sleep_url for bad in ("not-a-url", "http://portal.example.com/x", "/api/agents/i/sleep"): monkeypatch.setenv("GATEWAY_RELAY_SLEEP_URL", bad) assert brokered_sleep_url() is None, bad monkeypatch.setenv("GATEWAY_RELAY_SLEEP_URL", "https://portal.example.com/x?t=s") assert brokered_sleep_url() == "https://portal.example.com/x?t=s" def test_suspend_available_accepts_either_lever(monkeypatch): # No flaps socket but a broker exists, so the watcher may still quiesce. monkeypatch.setattr( "gateway.scale_to_zero.self_suspend_available", lambda *a, **k: False ) assert suspend_available({SLEEP_URL_ENV: _SLEEP_URL}) is True assert suspend_available({}) is False monkeypatch.setattr( "gateway.scale_to_zero.self_suspend_available", lambda *a, **k: True ) assert suspend_available({}) is True class _FakeResponse: def __init__(self, status): self.status = status def __enter__(self): return self def __exit__(self, *exc): return False def test_brokered_timeout_outlasts_the_broker_and_fits_the_redial_hold(): """Cross-repo deadline chain.""" from gateway.relay.ws_transport import REDIAL_HOLD_MAX_S from gateway.scale_to_zero import BROKERED_SUSPEND_TIMEOUT_S NAS_ROUTE_MAX_DURATION_S = 30.0 # mirrors the sleep route's maxDuration assert NAS_ROUTE_MAX_DURATION_S < BROKERED_SUSPEND_TIMEOUT_S assert BROKERED_SUSPEND_TIMEOUT_S < REDIAL_HOLD_MAX_S def test_request_brokered_suspend_posts_the_signed_url(): seen = {} def opener(request, timeout=None): seen["url"] = request.full_url seen["method"] = request.get_method() return _FakeResponse(200) assert request_brokered_suspend(_SLEEP_URL, opener=opener) is True assert seen == {"url": _SLEEP_URL, "method": "POST"} def test_request_brokered_suspend_fails_awake(): # Fail-awake: any refusal leaves the machine running rather than stranding a # frozen peer that still looks live to the connector. import urllib.error def http_error(request, timeout=None): raise urllib.error.HTTPError(_SLEEP_URL, 409, "Conflict", {}, None) def unreachable(request, timeout=None): raise urllib.error.URLError("connection refused") for opener in (http_error, unreachable, lambda *a, **k: _FakeResponse(500)): assert request_brokered_suspend(_SLEEP_URL, opener=opener) is False