feat(gateway): scale-to-zero idle detection + dormant-quiesce (Phase 0)

The gateway-side BEHAVIOUR layer that consumes the relay scale-to-zero
primitives (gateway-gateway Phase 5): the gateway decides it is idle and
drives the relay transport dormant so the platform (Fly autostop:"suspend")
can suspend the now-traffic-idle machine, which wakes on the connector's
wakeUrl poke (decisions.md Q3=C', D1-D13).

- gateway/scale_to_zero.py: pure helpers — scale_to_zero_enabled (the NAS
  Labs HERMES_SCALE_TO_ZERO stamp, D11/Q8=A), parse_idle_timeout_seconds
  (config.yaml gateway.scale_to_zero.idle_timeout_minutes, D2),
  messaging_is_relay_only_or_absent (F6/D1), should_arm (D1/D11/§3.4(1)),
  is_idle (D2/D3/F7).
- gateway/run.py: _last_inbound_at clock stamped on user inbound in
  _handle_message (F13); the arm-gate + idle predicate + the
  _scale_to_zero_watcher dormant sequence (mark draining -> adapter
  go_dormant() -> cooldown), started only when armed. Deliberately NOT the
  stop path and NOT mark_resume_pending (F12/D13).
- tools/process_registry.py: has_any_active() for the bg-work guard (D3/F7).
- hermes_cli/config.py: gateway.scale_to_zero.idle_timeout_minutes default 5.

Tests: 38 pure-logic + 6 watcher (incl. bg-work regression guard proven RED).
Full relay + scale-to-zero suites: 184 passed. The 20 unrelated failures in
the broader run are PRE-EXISTING on origin/main (custom-provider/tools tests),
confirmed via a pristine baseline worktree.

gateway/relay/adapter.py

@@ -341,6 +341,33 @@ class RelayAdapter(BasePlatformAdapter):
                     logger.debug("relay going_idle failed during drain", exc_info=True)
             await self._transport.disconnect()
 
+    async def go_dormant(self) -> bool:
+        """Quiesce the relay for a scale-to-zero suspend (D12 / Phase 0).
+
+        Unlike ``disconnect()`` (terminal teardown for shutdown/restart), this
+        keeps the adapter's reconnect path armed so the gateway re-dials and
+        drains its buffered backlog when the machine wakes. Delegates to the
+        transport's ``go_dormant()`` when available; a transport without it (the
+        stub) is a no-op that returns False, so callers degrade safely.
+
+        NOTE: deliberately does NOT stop the revocation monitor — going dormant
+        is not a teardown; the monitor stays live so a real opt-out/revocation
+        during dormancy is still surfaced on wake.
+        """
+        if self._transport is None:
+            return False
+        go_dormant = getattr(self._transport, "go_dormant", None)
+        if not callable(go_dormant):
+            return False
+        try:
+            result: Any = go_dormant()
+            if asyncio.iscoroutine(result):
+                return bool(await result)
+            return bool(result)
+        except Exception:  # noqa: BLE001 - dormancy is best-effort, never blocks the idle path
+            logger.debug("relay go_dormant failed", exc_info=True)
+            return False
+
     async def send(
         self,
         chat_id: str,

gateway/relay/ws_transport.py

@@ -232,6 +232,23 @@ class WebSocketRelayTransport:
         self._reconnect_backoff_s = reconnect_backoff_s
         self._reconnect_max_backoff_s = reconnect_max_backoff_s
         self._supervisor: Optional[asyncio.Task[None]] = None
+        # scale-to-zero §Phase 0 (D12/F14): a DORMANT close is distinct from both
+        # disconnect() (terminal: cancels the supervisor) and an unexpected close
+        # (re-dials immediately). go_dormant() sets this True, then closes the
+        # socket WITHOUT setting _closing — so _read_loop's fall-through still
+        # kicks the reconnect supervisor (the wake path stays armed), but the
+        # supervisor waits on the longer dormant cadence instead of the fast
+        # reconnect backoff, so it does not fight the platform's suspend window.
+        # On resume (process unfrozen) the pending wait completes, the re-dial
+        # succeeds, and the connector drains this instance's buffered backlog on
+        # the new handshake. Cleared on a successful re-dial (_dial_and_start).
+        self._dormant = False
+        # The re-dial poll cadence while dormant. A suspended machine's event
+        # loop is frozen, so this timer only advances once the machine is awake;
+        # it just needs to be short enough that a freshly-woken machine re-dials
+        # promptly (the connector's wake poke is what triggers the platform
+        # autostart in the first place — §3.4(5)).
+        self._dormant_redial_s = 1.0
 
         self._ws: Any = None
         self._reader: Optional[asyncio.Task[None]] = None
@@ -268,6 +285,10 @@ class WebSocketRelayTransport:
         # A fresh handshake is coming; clear any stale descriptor so handshake()
         # awaits the new one (matters on a re-dial).
         self._descriptor = None
+        # scale-to-zero (D12): a successful (re-)dial ends any dormant state — we
+        # are live again, so a subsequent UNEXPECTED close should reconnect on the
+        # normal fast backoff, not the dormant cadence.
+        self._dormant = False
         headers = self._upgrade_headers()
         if headers:
             self._ws = await websockets.connect(self._url, additional_headers=headers)  # type: ignore[union-attr]
@@ -389,6 +410,50 @@ class WebSocketRelayTransport:
         finally:
             self._going_idle_ack = None
 
+    async def go_dormant(self, timeout_s: float = 10.0) -> bool:
+        """Quiesce this transport for a scale-to-zero suspend (D12 / Phase 0).
+
+        Distinct from BOTH ``disconnect()`` and an unexpected close (F14):
+          - ``disconnect()`` sets ``_closing=True`` and CANCELS the reconnect
+            supervisor — terminal, "shutting down for good." A machine suspended
+            after that never re-dials on wake, so its buffered backlog strands.
+          - An unexpected close re-dials IMMEDIATELY (fast backoff) — the socket
+            never stays down, so the platform proxy never sees the connection go
+            away and never suspends the machine.
+
+        ``go_dormant()`` is the third mode the suspend behaviour needs:
+          1. ``go_idle()`` → the connector flips this instance to buffered-only
+             and acks (so inbound that arrives while we sleep buffers durably and
+             replays on the next handshake).
+          2. Close the socket so the platform proxy sees load drop to zero (the
+             precondition for Fly ``autostop:"suspend"``) — but WITHOUT setting
+             ``_closing``. The reader's normal end-of-socket fall-through still
+             arms the reconnect supervisor, so the wake path stays live; the
+             ``_dormant`` flag just makes that supervisor poll on the dormant
+             cadence rather than fight the suspend window.
+
+        On resume (process unfrozen) the supervisor's pending wait completes, the
+        re-dial succeeds, and the connector drains the buffered backlog on the new
+        handshake. Returns the ``go_idle`` ack result (True on ack); the dormancy
+        close happens regardless (a missed ack at worst races one live event onto
+        a closing socket, exactly as §5.3 already tolerates).
+
+        No-op-safe: a transport that never connected (``_ws is None``) just
+        returns False without closing.
+        """
+        if self._ws is None:
+            return False
+        acked = await self.go_idle(timeout_s=timeout_s)
+        # Mark dormant BEFORE closing so the supervisor (armed by the reader's
+        # fall-through) takes the dormant cadence, and a racing live event can't
+        # flip us back to a fast reconnect.
+        self._dormant = True
+        try:
+            await self._ws.close()
+        except Exception:  # noqa: BLE001 - best-effort; the reader still ends + arms reconnect
+            logger.debug("relay go_dormant: ws.close() raised", exc_info=True)
+        return acked
+
     async def _send_inbound_ack(self, buffer_id: str) -> None:
         """Acknowledge durable receipt of a buffered inbound delivery (§5.3).
 
@@ -489,8 +554,16 @@ class WebSocketRelayTransport:
         or disconnect() is called. NET-NEW for §5.3: a re-established socket makes
         the connector replay this instance's buffered-only backlog on the new
         handshake (the delivery-leg onResume). Never raises out (a re-dial failure
-        just retries); ends when a dial succeeds (its reader takes over) or closing."""
-        backoff = self._reconnect_backoff_s
+        just retries); ends when a dial succeeds (its reader takes over) or closing.
+
+        scale-to-zero (D12): when the close was a deliberate go_dormant() rather
+        than an unexpected drop, start from the dormant poll cadence. On a
+        suspended machine the event loop is frozen, so this sleep only advances
+        once the machine is awake — it just needs to be short enough that a
+        freshly-woken machine re-dials promptly. A successful _dial_and_start()
+        clears _dormant, so any LATER unexpected drop reconnects on the normal
+        fast backoff."""
+        backoff = self._dormant_redial_s if self._dormant else self._reconnect_backoff_s
         while not self._closing:
             try:
                 await asyncio.sleep(backoff)

gateway/run.py

@@ -2783,6 +2783,17 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         # Track background tasks to prevent garbage collection mid-execution
         self._background_tasks: set = set()
 
+        # scale-to-zero (Phase 0, F13): gateway-scoped "last inbound seen" clock.
+        # There is no such clock today (only a per-agent _last_activity_ts), so the
+        # idle predicate needs this. Stamped in _handle_message (the single inbound
+        # chokepoint all adapters call); seeded to "now" so a fresh gateway isn't
+        # considered idle from epoch. The scale-to-zero watcher (started only when
+        # the instance is opted in + relay-only + has a wakeUrl) reads it.
+        self._last_inbound_at: float = time.time()
+        # Set after a wake (re-arm cooldown, 0.F) so we don't immediately re-go
+        # dormant before the drained backlog has a chance to update the clock.
+        self._scale_to_zero_cooldown_until: float = 0.0
+
 
     def _wire_teams_pipeline_runtime(self) -> None:
         """Bind the Teams meeting pipeline runtime to Graph webhook ingress.
@@ -3568,6 +3579,145 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
     def _running_agent_count(self) -> int:
         return len(self._running_agents)
 
+    # ── scale-to-zero idle detection / dormant-quiesce (Phase 0) ──────────────
+    # The gateway-side BEHAVIOUR that consumes the relay scale-to-zero primitives
+    # (gateway-gateway Phase 5). Pure logic lives in gateway/scale_to_zero.py; the
+    # methods here bind it to the live runner/transport. See ~/nous/specs/
+    # scale-to-zero (decisions.md) for the design + the F12/F14 distinctions.
+
+    def _scale_to_zero_has_live_background_work(self) -> bool:
+        """Live background work that must block a suspend (D3/F7).
+
+        Backgrounded delegate_task / kanban / terminal(background=true) are NOT
+        counted by _running_agent_count(), but suspending mid-flight loses them.
+        Checks the runner's own tracked tasks + the process registry's running
+        processes + any pending process-completion watchers.
+        """
+        if any(not t.done() for t in self._background_tasks):
+            return True
+        try:
+            from tools.process_registry import process_registry
+
+            if process_registry.has_any_active():
+                return True
+            if process_registry.pending_watchers:
+                return True
+        except Exception:  # noqa: BLE001 - never let the idle check raise
+            logger.debug("scale-to-zero bg-work check failed", exc_info=True)
+        return False
+
+    def _scale_to_zero_idle_timeout_seconds(self) -> float:
+        from gateway.scale_to_zero import parse_idle_timeout_seconds
+
+        raw = None
+        try:
+            user_cfg = _load_gateway_config()
+            gw = user_cfg.get("gateway") if isinstance(user_cfg, dict) else None
+            stz = gw.get("scale_to_zero") if isinstance(gw, dict) else None
+            if isinstance(stz, dict):
+                raw = stz.get("idle_timeout_minutes")
+        except Exception:  # noqa: BLE001
+            raw = None
+        return parse_idle_timeout_seconds(raw)
+
+    def _scale_to_zero_should_arm(self) -> bool:
+        """Whether to start the idle watcher (D1/D11/§3.4(1))."""
+        from gateway.relay import relay_wake_url
+        from gateway.scale_to_zero import (
+            messaging_is_relay_only_or_absent,
+            scale_to_zero_enabled,
+            should_arm,
+        )
+
+        try:
+            platforms = list(self.config.platforms.keys()) if self.config else []
+        except Exception:  # noqa: BLE001
+            platforms = []
+        try:
+            wake_url = relay_wake_url()
+        except Exception:  # noqa: BLE001
+            wake_url = None
+        return should_arm(
+            enabled=scale_to_zero_enabled(),
+            relay_only_or_absent=messaging_is_relay_only_or_absent(platforms),
+            wake_url=wake_url,
+        )
+
+    def _scale_to_zero_is_idle(self) -> bool:
+        from gateway.scale_to_zero import is_idle
+
+        return is_idle(
+            running_agent_count=self._running_agent_count(),
+            seconds_since_last_inbound=time.time() - self._last_inbound_at,
+            idle_timeout_seconds=self._scale_to_zero_idle_timeout_seconds(),
+            has_live_background_work=self._scale_to_zero_has_live_background_work(),
+        )
+
+    def _relay_adapter_for_dormancy(self):
+        """Return the connected RELAY adapter, if any (the one go_dormant targets)."""
+        try:
+            from gateway.platforms.base import Platform
+        except Exception:  # noqa: BLE001
+            return None
+        return self.adapters.get(Platform.RELAY)
+
+    async def _scale_to_zero_watcher(self, interval: float = 30.0) -> None:
+        """Watch for idle and drive the relay dormant so the platform can suspend.
+
+        Started ONLY when _scale_to_zero_should_arm() (opted in via the Labs
+        HERMES_SCALE_TO_ZERO stamp + relay-only/absent messaging + a wakeUrl).
+        On a sustained idle window it runs the DORMANT sequence (D12/F12/F14):
+          - mark runtime status `draining` (composes with the existing state
+            machine, §3.4(6); does NOT set _running=False),
+          - relay adapter.go_dormant() — going_idle->ack + supervisor-preserving
+            socket close (NOT disconnect(), NOT the run.py stop path),
+          - deliberately NO mark_resume_pending (D13 — suspend preserves RAM).
+        The process stays alive; the platform (Fly autostop:"suspend") suspends
+        the now-traffic-idle machine and autostart wakes it on the wakeUrl poke,
+        at which point the preserved reconnect supervisor re-dials and the
+        connector drains the buffered backlog. After driving dormant we set a
+        re-arm cooldown so a wake's drained backlog isn't immediately re-quiesced.
+        """
+        await asyncio.sleep(min(interval, 30.0))  # let startup settle
+        while self._running:
+            try:
+                await asyncio.sleep(interval)
+                if not self._running:
+                    return
+                if time.time() < self._scale_to_zero_cooldown_until:
+                    continue
+                if not self._scale_to_zero_is_idle():
+                    continue
+                adapter = self._relay_adapter_for_dormancy()
+                if adapter is None:
+                    continue
+                go_dormant = getattr(adapter, "go_dormant", None)
+                if not callable(go_dormant):
+                    continue
+                logger.info(
+                    "scale-to-zero: gateway idle for >= %.0fs — going dormant "
+                    "(relay buffered, socket closed, awaiting platform suspend)",
+                    self._scale_to_zero_idle_timeout_seconds(),
+                )
+                try:
+                    self._update_runtime_status("draining")
+                except Exception:  # noqa: BLE001 - status is best-effort
+                    logger.debug("scale-to-zero: status mark failed", exc_info=True)
+                try:
+                    result = go_dormant()
+                    if asyncio.iscoroutine(result):
+                        await result
+                except Exception:  # noqa: BLE001 - dormancy is best-effort
+                    logger.debug("scale-to-zero: go_dormant failed", exc_info=True)
+                # 0.F: after a wake the drained inbound updates _last_inbound_at,
+                # but give it a window so we don't immediately re-go-dormant on the
+                # same idle reading before traffic lands.
+                self._scale_to_zero_cooldown_until = time.time() + max(interval, 60.0)
+            except asyncio.CancelledError:
+                raise
+            except Exception:  # noqa: BLE001 - the watcher must never crash the gateway
+                logger.debug("scale-to-zero watcher iteration error", exc_info=True)
+
     def _status_action_label(self) -> str:
         return "restart" if self._restart_requested else "shutdown"
 
@@ -5965,6 +6115,22 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         # idle case where the subagent finishes with no agent turn running.
         asyncio.create_task(self._async_delegation_watcher())
 
+        # Start the scale-to-zero idle watcher ONLY when this instance is opted
+        # in (the NAS "Labs" HERMES_SCALE_TO_ZERO stamp), messaging is
+        # relay-only/absent, and a wakeUrl is registered (decisions.md D1/D11/
+        # §3.4(1)). A non-opted instance never starts it, so behaviour is exactly
+        # as today. When armed, the watcher drives the relay dormant on sustained
+        # idle so the platform (Fly autostop:"suspend") can suspend the machine.
+        try:
+            if self._scale_to_zero_should_arm():
+                logger.info(
+                    "scale-to-zero: armed (idle timeout %.0fs) — watching for idle",
+                    self._scale_to_zero_idle_timeout_seconds(),
+                )
+                asyncio.create_task(self._scale_to_zero_watcher())
+        except Exception:  # noqa: BLE001 - arming must never block startup
+            logger.debug("scale-to-zero: arm check failed at startup", exc_info=True)
+
         logger.info("Press Ctrl+C to stop")
         
         return True
@@ -7332,6 +7498,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         # are system-generated and must skip user authorization.
         is_internal = bool(getattr(event, "internal", False))
 
+        # scale-to-zero (Phase 0, 0.B/F13): stamp the gateway-scoped last-inbound
+        # clock for real (user-originated) inbound only. Internal/system events
+        # (background-process completions, startup-restore replays) are NOT
+        # traffic — counting them would keep a genuinely idle gateway awake. This
+        # clock is what the idle predicate (gateway/scale_to_zero.is_idle) reads.
+        if not is_internal:
+            self._last_inbound_at = time.time()
+
         # Fire pre_gateway_dispatch plugin hook for user-originated messages.
         # Plugins receive the MessageEvent and may return a dict influencing flow:
         #   {"action": "skip",    "reason": ...}    -> drop (no reply, plugin handled)

gateway/scale_to_zero.py

@@ -0,0 +1,124 @@
+"""Scale-to-zero idle detection + dormant-quiesce for the gateway (Phase 0).
+
+This is the gateway-side BEHAVIOUR layer that consumes the relay scale-to-zero
+PRIMITIVES (gateway-gateway Phase 5: the buffered-flip, the durable per-instance
+buffer, the wakeUrl poke, the reconnect supervisor). It owns the *decision* to go
+idle and drives the relay transport's ``go_dormant()`` (D12) — it does NOT itself
+suspend the machine. On Fly, the now-traffic-idle machine is suspended by
+``autostop:"suspend"`` and woken by autostart-on-wakeUrl (decisions.md Q3=C′).
+
+Design constraints (decisions.md):
+  - Per-instance enable is gated SOLELY by the NAS "Labs" toggle, carried to the
+    gateway as the ``HERMES_SCALE_TO_ZERO`` env stamp (D11/Q8=A). NOT a user
+    config key; ``scale_to_zero.idle_timeout_minutes`` IS config.yaml (D2).
+  - Arm only when messaging is relay-only or absent (D1/F6) AND a wakeUrl is
+    registered (§3.4(1)) AND the flag is set.
+  - Idle = no in-flight agent turn AND no inbound for N min AND no live
+    background work (D2/D3/F7).
+  - The quiesce uses ``go_dormant()`` (socket closed + supervisor preserved),
+    NEVER the stop/restart drain or ``disconnect()`` (F12/F14). The process stays
+    alive; Fly freezes+resumes it.
+  - ``mark_resume_pending`` is deliberately NOT called here (D13 — suspend
+    preserves RAM; revive only if we move to autostop:"stop" or see kills).
+
+The pure helpers (``parse_idle_timeout_seconds``, ``scale_to_zero_enabled``,
+``messaging_is_relay_only_or_absent``, ``is_idle``, ``should_arm``) take plain
+inputs so they unit-test without a live gateway.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any, Iterable, Optional
+
+# Env flag stamped by NAS when the scaleToZero Labs toggle is on (D11/Q8=A),
+# mirroring how the `relay` feature stamps GATEWAY_RELAY_URL. Truthy values only.
+SCALE_TO_ZERO_ENV = "HERMES_SCALE_TO_ZERO"
+
+# config.yaml default (D2). Behavioural setting -> config, not env.
+DEFAULT_IDLE_TIMEOUT_MINUTES = 5
+
+_TRUTHY = {"1", "true", "yes", "on"}
+
+
+def scale_to_zero_enabled(environ: Optional[dict] = None) -> bool:
+    """Whether the per-instance Labs toggle is on (the HERMES_SCALE_TO_ZERO stamp).
+
+    D11/Q8=A: this env flag is the SOLE per-instance enable signal reaching the
+    gateway. Absent/blank/falsey -> disabled (fail-safe default off).
+    """
+    env = environ if environ is not None else os.environ
+    return str(env.get(SCALE_TO_ZERO_ENV, "")).strip().lower() in _TRUTHY
+
+
+def parse_idle_timeout_seconds(
+    cfg_value: Any, default_minutes: int = DEFAULT_IDLE_TIMEOUT_MINUTES
+) -> float:
+    """Coerce ``scale_to_zero.idle_timeout_minutes`` (config.yaml, D2) to seconds.
+
+    Degrades to the default on any non-numeric / non-positive value (never raises,
+    never returns <= 0 — a zero/negative timeout would make the gateway go dormant
+    instantly, which is never the intent).
+    """
+    try:
+        minutes = float(cfg_value)
+    except (TypeError, ValueError):
+        minutes = float(default_minutes)
+    if minutes <= 0:
+        minutes = float(default_minutes)
+    return minutes * 60.0
+
+
+def messaging_is_relay_only_or_absent(platforms: Iterable[Any]) -> bool:
+    """True iff the only connected messaging platform is RELAY, or there is none
+    (a Chronos-only / no-platform agent) — the F6/D1 structural precondition.
+
+    A directly-connected platform (Discord/Telegram/Slack/...) holds a live
+    socket and cannot scale to zero, so its presence disarms the feature. We
+    compare by the platform's ``.value``/name to avoid importing the enum here
+    (keeps this module import-light and unit-testable).
+    """
+    names = {_platform_name(p) for p in platforms}
+    names.discard("relay")
+    return len(names) == 0
+
+
+def _platform_name(platform: Any) -> str:
+    value = getattr(platform, "value", platform)
+    return str(value).strip().lower()
+
+
+def should_arm(
+    *,
+    enabled: bool,
+    relay_only_or_absent: bool,
+    wake_url: Optional[str],
+) -> bool:
+    """Whether to start the idle watcher at all (D1/D11/§3.4(1)).
+
+    ALL must hold: the Labs flag is on, messaging is relay-only/absent, and a
+    wakeUrl is registered (a suspended instance with no reachable wake target is
+    a black hole — §3.4(1)). Any unmet -> the watcher never starts (no idle
+    timer, no dormancy), so a non-opted instance behaves exactly as today.
+    """
+    return bool(enabled) and bool(relay_only_or_absent) and bool(wake_url)
+
+
+def is_idle(
+    *,
+    running_agent_count: int,
+    seconds_since_last_inbound: float,
+    idle_timeout_seconds: float,
+    has_live_background_work: bool,
+) -> bool:
+    """The idle predicate (D2/D3/F7). Pure — composes the three conjuncts.
+
+    Idle iff: no in-flight agent turn, no inbound within the timeout window, and
+    no live background work (backgrounded delegate_task / kanban / bg terminal).
+    Any active work keeps the gateway awake — suspending mid-flight would lose it.
+    """
+    if running_agent_count > 0:
+        return False
+    if has_live_background_work:
+        return False
+    return seconds_since_last_inbound >= idle_timeout_seconds

hermes_cli/config.py

@@ -2522,6 +2522,18 @@ DEFAULT_CONFIG = {
     # Gateway settings — control how messaging platforms (Telegram, Discord,
     # Slack, etc.) deliver agent-produced files as native attachments.
     "gateway": {
+        # Scale-to-zero idle detection (Phase 0). The gateway watches for idle
+        # and, when an instance is opted in via the NAS "Labs" toggle (carried as
+        # the HERMES_SCALE_TO_ZERO env stamp) AND messaging is relay-only/absent
+        # AND a wakeUrl is registered, drives the relay transport dormant so the
+        # platform (e.g. Fly autostop:"suspend") can suspend the now-idle machine;
+        # it wakes on the connector's wakeUrl poke. This is the idle TIMEOUT only
+        # — whether the feature is enabled at all is the Labs toggle, never a
+        # config key (decisions.md D2/D11). 0/negative falls back to the default.
+        "scale_to_zero": {
+            "idle_timeout_minutes": 5,
+        },
+
         # Inject a human-readable timestamp prefix (e.g.
         # "[Tue 2026-04-28 13:40:53 CEST]") onto user messages IN THE MODEL'S
         # CONTEXT so the agent has temporal awareness of when each message was

package-lock.json

@@ -1419,7 +1419,6 @@
       "os": [
         "aix"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1436,7 +1435,6 @@
       "os": [
         "android"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1453,7 +1451,6 @@
       "os": [
         "android"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1470,7 +1467,6 @@
       "os": [
         "android"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1487,7 +1483,6 @@
       "os": [
         "darwin"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1504,7 +1499,6 @@
       "os": [
         "darwin"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1521,7 +1515,6 @@
       "os": [
         "freebsd"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1538,7 +1531,6 @@
       "os": [
         "freebsd"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1555,7 +1547,6 @@
       "os": [
         "linux"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1572,7 +1563,6 @@
       "os": [
         "linux"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1589,7 +1579,6 @@
       "os": [
         "linux"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1606,7 +1595,6 @@
       "os": [
         "linux"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1623,7 +1611,6 @@
       "os": [
         "linux"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1640,7 +1627,6 @@
       "os": [
         "linux"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1657,7 +1643,6 @@
       "os": [
         "linux"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1674,7 +1659,6 @@
       "os": [
         "linux"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1691,7 +1675,6 @@
       "os": [
         "linux"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1708,7 +1691,6 @@
       "os": [
         "netbsd"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1725,7 +1707,6 @@
       "os": [
         "netbsd"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1742,7 +1723,6 @@
       "os": [
         "openbsd"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1759,7 +1739,6 @@
       "os": [
         "openbsd"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1776,7 +1755,6 @@
       "os": [
         "openharmony"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1793,7 +1771,6 @@
       "os": [
         "sunos"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1810,7 +1787,6 @@
       "os": [
         "win32"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1827,7 +1803,6 @@
       "os": [
         "win32"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1844,7 +1819,6 @@
       "os": [
         "win32"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }

tests/gateway/relay/test_relay_going_idle.py

@@ -241,3 +241,195 @@ async def test_adapter_emits_going_idle_on_disconnect(server):
     await adapter.connect()
     await adapter.disconnect()
     assert server.going_idle_count == 1
+
+
+# ── scale-to-zero go_dormant() (D12 / F14) ───────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_go_dormant_emits_going_idle_and_closes_without_terminal_teardown(server):
+    """go_dormant() flips the connector to buffered-only (going_idle->ack) AND
+    closes the socket, but does NOT set the terminal _closing flag or cancel the
+    reconnect supervisor — the F14 distinction from disconnect()."""
+    t = WebSocketRelayTransport(
+        server.url, "discord", "appShared", reconnect=True, reconnect_backoff_s=0.05
+    )
+    await t.connect()
+    await t.handshake()
+    try:
+        acked = await t.go_dormant(timeout_s=2)
+        assert acked is True
+        assert server.going_idle_count == 1
+        # The socket was closed (dormant), but NOT via the terminal path:
+        assert t._closing is False  # disconnect() would set this True
+        assert t._dormant is True
+        # Not a revocation — the auth-revoked latch stays clear.
+        assert t.auth_revoked is False
+    finally:
+        await t.disconnect()
+
+
+@pytest.mark.asyncio
+async def test_go_dormant_redials_on_wake_and_drains(server):
+    """After go_dormant() the reconnect supervisor stays armed, so the gateway
+    re-dials (simulating a wake) and the connector replays its buffered backlog
+    on the new handshake. This is the wake->reconnect->drain contract (§3.4)."""
+    # Queue a buffered inbound to be replayed on the NEXT (wake) handshake.
+    server._to_push = [
+        {
+            "type": "inbound",
+            "event": {
+                "text": "while-asleep",
+                "message_type": "text",
+                "source": {"platform": "discord", "chat_id": "c1", "chat_type": "dm"},
+            },
+            "bufferId": "buf-wake-1",
+        }
+    ]
+    seen: list[str] = []
+
+    async def handler(ev):
+        seen.append(ev.text)
+
+    t = WebSocketRelayTransport(
+        server.url, "discord", "appShared", reconnect=True, reconnect_backoff_s=5.0
+    )
+    # Dormant re-dial cadence is short so the test wakes promptly even though the
+    # ordinary reconnect backoff is long (proves the dormant path uses its own).
+    t._dormant_redial_s = 0.05
+    t.set_inbound_handler(handler)
+    await t.connect()
+    await t.handshake()
+    before = server.connections
+    try:
+        await t.go_dormant(timeout_s=2)
+        # The supervisor was armed by the dormant close; it re-dials on the
+        # dormant cadence (~0.05s), NOT the 5s reconnect backoff.
+        for _ in range(50):
+            if server.connections > before and "while-asleep" in seen:
+                break
+            await asyncio.sleep(0.05)
+        assert server.connections > before  # re-dialed (woke)
+        assert "while-asleep" in seen  # drained the buffered backlog on reconnect
+        # The successful re-dial cleared the dormant flag.
+        assert t._dormant is False
+        # The buffered entry was acked (this stub re-pushes on every handshake, so
+        # a long-lived dormant poll may ack it more than once; the invariant is
+        # that it was drained at least once — a real connector stops replaying an
+        # acked entry).
+        assert "buf-wake-1" in server.inbound_acks
+    finally:
+        await t.disconnect()
+
+
+@pytest.mark.asyncio
+async def test_disconnect_cancels_supervisor_but_go_dormant_does_not(server):
+    """Direct contrast (F14): disconnect() is terminal (cancels supervisor, no
+    re-dial); go_dormant() keeps it armed. Guards against a future refactor that
+    routes dormancy through disconnect()."""
+    # disconnect(): terminal — no reconnect.
+    t1 = WebSocketRelayTransport(
+        server.url, "discord", "appShared", reconnect=True, reconnect_backoff_s=0.05
+    )
+    await t1.connect()
+    await t1.handshake()
+    after_first = server.connections
+    await t1.disconnect()
+    await asyncio.sleep(0.3)
+    assert server.connections == after_first  # disconnect did NOT re-dial
+    assert t1._closing is True
+
+    # go_dormant(): armed — re-dials.
+    t2 = WebSocketRelayTransport(
+        server.url, "discord", "appShared", reconnect=True, reconnect_backoff_s=0.05
+    )
+    t2._dormant_redial_s = 0.05
+    await t2.connect()
+    await t2.handshake()
+    before = server.connections
+    try:
+        await t2.go_dormant(timeout_s=2)
+        for _ in range(50):
+            if server.connections > before:
+                break
+            await asyncio.sleep(0.05)
+        assert server.connections > before  # go_dormant stayed armed and re-dialed
+        assert t2._closing is False
+    finally:
+        await t2.disconnect()
+
+
+@pytest.mark.asyncio
+async def test_go_dormant_noop_when_never_connected():
+    """go_dormant() on a transport that never connected is a safe no-op (False),
+    not a crash."""
+    t = WebSocketRelayTransport("ws://127.0.0.1:1", "discord", "appShared")
+    assert await t.go_dormant(timeout_s=0.1) is False
+
+
+@pytest.mark.asyncio
+async def test_adapter_go_dormant_delegates_to_transport(server):
+    """RelayAdapter.go_dormant() drives the transport's go_dormant (going_idle +
+    dormant close) without the terminal teardown disconnect() does."""
+    from gateway.config import PlatformConfig
+    from gateway.relay.adapter import RelayAdapter
+    from gateway.relay.descriptor import CONTRACT_VERSION, CapabilityDescriptor
+
+    placeholder = CapabilityDescriptor(
+        contract_version=CONTRACT_VERSION,
+        platform="discord",
+        label="Relay",
+        max_message_length=4096,
+        supports_draft_streaming=False,
+        supports_edit=True,
+        supports_threads=False,
+        markdown_dialect="plain",
+        len_unit="chars",
+    )
+    transport = WebSocketRelayTransport(
+        server.url, "discord", "appShared", reconnect=True, reconnect_backoff_s=0.05
+    )
+    adapter = RelayAdapter(PlatformConfig(), placeholder, transport=transport)
+    await adapter.connect()
+    try:
+        ok = await adapter.go_dormant()
+        assert ok is True
+        assert server.going_idle_count == 1
+        assert transport._closing is False  # NOT the terminal teardown
+        assert transport._dormant is True
+    finally:
+        await adapter.disconnect()
+
+
+@pytest.mark.asyncio
+async def test_adapter_go_dormant_noop_on_stub_transport():
+    """An adapter whose transport lacks go_dormant (the stub) degrades to a safe
+    no-op returning False, never raising."""
+    from gateway.config import PlatformConfig
+    from gateway.relay.adapter import RelayAdapter
+    from gateway.relay.descriptor import CONTRACT_VERSION, CapabilityDescriptor
+
+    placeholder = CapabilityDescriptor(
+        contract_version=CONTRACT_VERSION,
+        platform="discord",
+        label="Relay",
+        max_message_length=4096,
+        supports_draft_streaming=False,
+        supports_edit=True,
+        supports_threads=False,
+        markdown_dialect="plain",
+        len_unit="chars",
+    )
+
+    class _StubTransport:
+        async def connect(self):
+            return True
+
+        def set_inbound_handler(self, h):
+            pass
+
+        async def handshake(self):
+            return placeholder
+
+    adapter = RelayAdapter(PlatformConfig(), placeholder, transport=_StubTransport())
+    assert await adapter.go_dormant() is False

tests/gateway/test_scale_to_zero.py

@@ -0,0 +1,144 @@
+"""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
+
+
+@pytest.mark.parametrize("value", ["", "0", "false", "no", "off", "nope"])
+def test_enabled_falsey_values(value):
+    assert scale_to_zero_enabled({SCALE_TO_ZERO_ENV: value}) is False
+
+
+def test_enabled_absent_is_false():
+    # Fail-safe default OFF when the stamp is absent (a non-opted instance).
+    assert scale_to_zero_enabled({}) is False
+
+
+# ── 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
+
+
+@pytest.mark.parametrize("bad", [None, "", "abc", {}, [], object()])
+def test_timeout_degrades_to_default_on_garbage(bad):
+    assert parse_idle_timeout_seconds(bad) == DEFAULT_IDLE_TIMEOUT_MINUTES * 60.0
+
+
+@pytest.mark.parametrize("nonpos", [0, -1, -30, "0", "-5"])
+def test_timeout_rejects_nonpositive(nonpos):
+    # A zero/negative timeout would go dormant instantly — never the intent.
+    assert parse_idle_timeout_seconds(nonpos) == 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_direct_socket_platform_disarms():
+    assert messaging_is_relay_only_or_absent([_P("discord")]) is False
+    assert messaging_is_relay_only_or_absent([_P("relay"), _P("telegram")]) is False
+
+
+def test_accepts_bare_strings_too():
+    assert messaging_is_relay_only_or_absent(["relay"]) is True
+    assert messaging_is_relay_only_or_absent(["discord"]) is False
+
+
+# ── should_arm (D1/D11/§3.4(1)) ──────────────────────────────────────────────
+
+
+def test_arm_requires_all_three():
+    assert should_arm(enabled=True, relay_only_or_absent=True, wake_url="https://x") is True
+
+
+def test_arm_blocked_when_flag_off():
+    assert should_arm(enabled=False, relay_only_or_absent=True, wake_url="https://x") is False
+
+
+def test_arm_blocked_when_direct_socket():
+    assert should_arm(enabled=True, relay_only_or_absent=False, wake_url="https://x") is False
+
+
+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(
+        running_agent_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_idle_true_when_all_quiet():
+    assert is_idle(**_idle_kwargs()) is True
+
+
+def test_not_idle_with_running_agent():
+    assert is_idle(**_idle_kwargs(running_agent_count=1)) is False
+
+
+def test_not_idle_within_timeout_window():
+    assert is_idle(**_idle_kwargs(seconds_since_last_inbound=120.0)) 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
+
+
+def test_not_idle_with_live_background_work():
+    assert is_idle(**_idle_kwargs(has_live_background_work=True)) is False

tests/gateway/test_scale_to_zero_watcher.py

@@ -0,0 +1,120 @@
+"""Watcher-level tests for scale-to-zero: the idle watcher's dormant sequence and
+the arm-gate wiring, exercised against the real GatewayRunner methods bound onto
+a lightweight stand-in (booting a full gateway is unnecessary for this logic and
+would be slow/flaky).
+
+These cover the parts gateway/test_scale_to_zero.py (pure helpers) can't: that
+the watcher calls the relay adapter's go_dormant() exactly when idle+armed,
+respects the cooldown, and skips when busy — the F7/D3 + D12 behaviour.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import time
+
+import pytest
+
+from gateway.run import GatewayRunner
+
+
+class _FakeRelayAdapter:
+    def __init__(self):
+        self.go_dormant_calls = 0
+
+    async def go_dormant(self):
+        self.go_dormant_calls += 1
+        return True
+
+
+def _runner_with(monkeypatch, *, idle, armed_adapter=True):
+    """Build a GatewayRunner without booting it, stubbing just what the watcher
+    touches. Real methods (_scale_to_zero_is_idle composition, the watcher body)
+    run; only their dependencies are stubbed."""
+    r = GatewayRunner.__new__(GatewayRunner)
+    r._running = True
+    r._scale_to_zero_cooldown_until = 0.0
+    r._last_inbound_at = time.time()
+    r._running_agents = {}
+    r._background_tasks = set()
+    adapter = _FakeRelayAdapter() if armed_adapter else None
+
+    monkeypatch.setattr(r, "_scale_to_zero_is_idle", lambda: idle, raising=False)
+    monkeypatch.setattr(r, "_relay_adapter_for_dormancy", lambda: adapter, raising=False)
+    monkeypatch.setattr(r, "_scale_to_zero_idle_timeout_seconds", lambda: 300.0, raising=False)
+    monkeypatch.setattr(r, "_update_runtime_status", lambda *a, **k: None, raising=False)
+    return r, adapter
+
+
+@pytest.mark.asyncio
+async def test_watcher_goes_dormant_when_idle(monkeypatch):
+    r, adapter = _runner_with(monkeypatch, idle=True)
+    # Run one iteration: stop after the first sleep so the loop exits cleanly.
+    task = asyncio.create_task(r._scale_to_zero_watcher(interval=0.01))
+    await asyncio.sleep(0.1)
+    r._running = False
+    await asyncio.wait_for(task, timeout=2)
+    assert adapter.go_dormant_calls >= 1
+    # After driving dormant, a re-arm cooldown is set (0.F).
+    assert r._scale_to_zero_cooldown_until > time.time()
+
+
+@pytest.mark.asyncio
+async def test_watcher_does_not_go_dormant_when_busy(monkeypatch):
+    r, adapter = _runner_with(monkeypatch, idle=False)
+    task = asyncio.create_task(r._scale_to_zero_watcher(interval=0.01))
+    await asyncio.sleep(0.1)
+    r._running = False
+    await asyncio.wait_for(task, timeout=2)
+    assert adapter.go_dormant_calls == 0
+
+
+@pytest.mark.asyncio
+async def test_watcher_respects_cooldown(monkeypatch):
+    r, adapter = _runner_with(monkeypatch, idle=True)
+    # Cooldown active far in the future: even though idle, no dormancy fires.
+    r._scale_to_zero_cooldown_until = time.time() + 3600
+    task = asyncio.create_task(r._scale_to_zero_watcher(interval=0.01))
+    await asyncio.sleep(0.1)
+    r._running = False
+    await asyncio.wait_for(task, timeout=2)
+    assert adapter.go_dormant_calls == 0
+
+
+@pytest.mark.asyncio
+async def test_watcher_noop_when_no_relay_adapter(monkeypatch):
+    # Armed-but-no-relay-adapter (e.g. relay not yet connected): must not crash.
+    r, _ = _runner_with(monkeypatch, idle=True, armed_adapter=False)
+    task = asyncio.create_task(r._scale_to_zero_watcher(interval=0.01))
+    await asyncio.sleep(0.1)
+    r._running = False
+    await asyncio.wait_for(task, timeout=2)
+    # No exception, loop exits cleanly — nothing to assert beyond survival.
+
+
+def test_bg_work_blocks_idle_via_background_tasks(monkeypatch):
+    """_scale_to_zero_has_live_background_work() reports True when a tracked
+    background task is still live (D3/F7) — the guard that keeps a gateway with
+    an in-flight backgrounded subagent/terminal awake."""
+    r = GatewayRunner.__new__(GatewayRunner)
+
+    async def _never():
+        await asyncio.sleep(3600)
+
+    loop = asyncio.new_event_loop()
+    try:
+        t = loop.create_task(_never())
+        r._background_tasks = {t}
+        # process_registry has nothing active in this fresh process.
+        assert r._scale_to_zero_has_live_background_work() is True
+        t.cancel()
+    finally:
+        loop.run_until_complete(asyncio.gather(t, return_exceptions=True))
+        loop.close()
+
+
+def test_bg_work_false_when_quiet():
+    r = GatewayRunner.__new__(GatewayRunner)
+    r._background_tasks = set()
+    # No background tasks, no active processes in this fresh process.
+    assert r._scale_to_zero_has_live_background_work() is False

tools/process_registry.py

@@ -1581,6 +1581,23 @@ class ProcessRegistry:
                 for s in self._running.values()
             )
 
+    def has_any_active(self) -> bool:
+        """Whether ANY background process is still running (across all sessions).
+
+        Used by scale-to-zero idle detection (gateway/scale_to_zero): a gateway
+        with a live background process (terminal background=true) is NOT idle and
+        must not be suspended, or the process is lost. Refreshes detached
+        sessions first so a finished-but-unreaped process reads as inactive.
+        """
+        with self._lock:
+            sessions = list(self._running.values())
+
+        for session in sessions:
+            self._refresh_detached_session(session)
+
+        with self._lock:
+            return any(not s.exited for s in self._running.values())
+
     def kill_all(self, task_id: str = None) -> int:
         """Kill all running processes, optionally filtered by task_id. Returns count killed."""
         with self._lock: