feat(cli): global emergency stop — `hermes pause` / `hermes resume`
Resumable ESTOP sentinel at $HERMES_HOME/ESTOP that halts NEW work only: - agent/estop.py: sentinel engage/disengage/is_engaged (single stat, no caching), optional reason + timestamp stored as JSON, paused_reply() notice, check_paused() log-once-per-engagement helper. Corrupt/empty sentinel still pauses (fail safe); a `touch ~/.hermes/ESTOP` works. - cron/scheduler.py: tick() skips dispatch while engaged (logged once per engagement, not per tick). Due jobs simply wait for the next tick after resume — in-flight runs are never touched. - gateway/kanban_watchers.py: dispatcher skips auto-decompose and worker spawning while engaged; zombie reaping still runs and running workers finish naturally. - gateway/run.py: new gateway turns (post-auth, non-internal) get a brief "Hermes is paused" reply instead of an agent run. Internal events (in-flight background completions) bypass the gate. - hermes_cli/subcommands/pause.py: `hermes pause [--reason]` and `hermes resume`, wired into main() and _BUILTIN_SUBCOMMANDS. - hermes_cli/status.py: `hermes status` shows a PAUSED banner (one stat). - tests/test_estop.py: 20 tests — sentinel lifecycle, reason surfacing, log-once, cron skip + resume, kanban gate, gateway paused reply + internal bypass, CLI idempotence, builtin-set parity, status line. Never kills in-flight work; resumable with no restart. Footprint ladder: CLI command only, no new model tool, no new env vars. Ported from: gastownhall/gastown estop.go (MIT); related prior art: #26778 (/panic — kill/exit semantics, deliberately different: ours is a resumable pause), #44617 (interrupt in-flight cron — out of scope here).
This commit is contained in:
Teknium
parent
5396da844a
commit
5db1b72b1f
|
|
@ -0,0 +1,167 @@
|
|||
"""Global emergency stop (ESTOP) — a resumable pause for NEW work only.
|
||||
|
||||
``hermes pause`` writes a sentinel file at ``$HERMES_HOME/ESTOP``;
|
||||
``hermes resume`` removes it. While the sentinel exists:
|
||||
|
||||
* the cron scheduler skips dispatching due jobs (``cron/scheduler.py:tick``),
|
||||
* the embedded kanban dispatcher skips spawning workers
|
||||
(``gateway/kanban_watchers.py``),
|
||||
* new gateway turns get a brief "Hermes is paused" reply instead of an
|
||||
agent run (``gateway/run.py:_handle_message``).
|
||||
|
||||
In-flight work is NEVER killed — this is pause-new-work, not panic/exit.
|
||||
The check is a single ``os.stat`` so callers may run it every tick; no
|
||||
caching beyond the OS is performed, so engaging/disengaging takes effect on
|
||||
the very next check.
|
||||
|
||||
The sentinel body is optional JSON ``{"reason": ..., "engaged_at": ...}``.
|
||||
A corrupt or empty file still counts as engaged (fail safe): the pause must
|
||||
hold even if the file was created by ``touch ~/.hermes/ESTOP``.
|
||||
|
||||
Ported from: gastownhall/gastown estop.go (MIT). Related prior art:
|
||||
#26778 (/panic — kill/exit semantics; deliberately different, ours is
|
||||
resumable) and #44617 (interrupting in-flight cron; deliberately out of
|
||||
scope here).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import threading
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
SENTINEL_NAME = "ESTOP"
|
||||
|
||||
# Per-component "logged already for this engagement" flags so a paused
|
||||
# dispatch loop logs once per engagement instead of once per tick.
|
||||
_log_lock = threading.Lock()
|
||||
_logged_components: set[str] = set()
|
||||
|
||||
|
||||
def _hermes_home() -> Path:
|
||||
"""Resolve the active HERMES_HOME (profile-aware) at call time."""
|
||||
try:
|
||||
from hermes_constants import get_hermes_home
|
||||
return get_hermes_home()
|
||||
except Exception:
|
||||
return Path(os.path.expanduser("~/.hermes"))
|
||||
|
||||
|
||||
def sentinel_path() -> Path:
|
||||
"""Path of the ESTOP sentinel under the active HERMES_HOME."""
|
||||
return _hermes_home() / SENTINEL_NAME
|
||||
|
||||
|
||||
def is_engaged() -> bool:
|
||||
"""Cheap check (one stat): is the global emergency stop engaged?"""
|
||||
try:
|
||||
return sentinel_path().exists()
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
|
||||
def engage(reason: Optional[str] = None) -> Path:
|
||||
"""Create the ESTOP sentinel. Idempotent; re-engaging updates the file."""
|
||||
path = sentinel_path()
|
||||
payload = {
|
||||
"engaged_at": datetime.now(timezone.utc).isoformat(),
|
||||
"reason": reason or None,
|
||||
}
|
||||
try:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
|
||||
except OSError:
|
||||
# Best effort: an empty/partial sentinel still pauses (fail safe).
|
||||
try:
|
||||
path.touch(exist_ok=True)
|
||||
except OSError:
|
||||
pass
|
||||
return path
|
||||
|
||||
|
||||
def disengage() -> bool:
|
||||
"""Remove the ESTOP sentinel. Returns True if a pause was lifted."""
|
||||
try:
|
||||
sentinel_path().unlink()
|
||||
return True
|
||||
except FileNotFoundError:
|
||||
return False
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
|
||||
def get_state() -> Optional[dict]:
|
||||
"""Return ``{"reason": ..., "engaged_at": ...}`` or None when not engaged.
|
||||
|
||||
A sentinel with an unreadable/corrupt body still reports engaged, with
|
||||
both fields None — the pause is authoritative, the metadata is not.
|
||||
"""
|
||||
path = sentinel_path()
|
||||
if not path.exists():
|
||||
return None
|
||||
reason = None
|
||||
engaged_at = None
|
||||
try:
|
||||
raw = json.loads(path.read_text(encoding="utf-8"))
|
||||
if isinstance(raw, dict):
|
||||
reason = raw.get("reason") or None
|
||||
engaged_at = raw.get("engaged_at") or None
|
||||
except (OSError, ValueError):
|
||||
pass
|
||||
return {"reason": reason, "engaged_at": engaged_at}
|
||||
|
||||
|
||||
def paused_reply() -> Optional[str]:
|
||||
"""Short user-facing notice for new gateway turns, or None if not paused."""
|
||||
state = get_state()
|
||||
if state is None:
|
||||
return None
|
||||
reason = state.get("reason")
|
||||
if reason:
|
||||
return (
|
||||
f"⏸️ Hermes is paused ({reason}). New work is on hold; "
|
||||
"run `hermes resume` to pick things back up."
|
||||
)
|
||||
return (
|
||||
"⏸️ Hermes is paused. New work is on hold; "
|
||||
"run `hermes resume` to pick things back up."
|
||||
)
|
||||
|
||||
|
||||
def check_paused(component: str, logger: logging.Logger) -> bool:
|
||||
"""Return True when engaged, logging once per engagement per component.
|
||||
|
||||
Dispatch loops call this every tick; the log fires on the disengaged→
|
||||
engaged transition for that component and re-arms after a resume, so a
|
||||
long pause doesn't spam one line per tick.
|
||||
"""
|
||||
if not is_engaged():
|
||||
with _log_lock:
|
||||
_logged_components.discard(component)
|
||||
return False
|
||||
with _log_lock:
|
||||
first = component not in _logged_components
|
||||
if first:
|
||||
_logged_components.add(component)
|
||||
if first:
|
||||
state = get_state() or {}
|
||||
reason = state.get("reason")
|
||||
suffix = f" (reason: {reason})" if reason else ""
|
||||
logger.info(
|
||||
"%s dispatch paused by global emergency stop%s — remove with "
|
||||
"`hermes resume` (%s)",
|
||||
component,
|
||||
suffix,
|
||||
sentinel_path(),
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
def _reset_log_state_for_tests() -> None:
|
||||
"""Clear the log-once bookkeeping (test isolation helper)."""
|
||||
with _log_lock:
|
||||
_logged_components.clear()
|
||||
|
|
@ -4759,6 +4759,17 @@ def tick(
|
|||
return 0
|
||||
|
||||
try:
|
||||
# Global emergency stop (`hermes pause`): skip dispatch entirely while
|
||||
# the ESTOP sentinel exists. Never touches in-flight runs — due jobs
|
||||
# simply wait for the next tick after `hermes resume`. Logged once per
|
||||
# engagement (not every tick) by check_paused.
|
||||
try:
|
||||
from agent.estop import check_paused as _estop_check_paused
|
||||
if _estop_check_paused("cron", logger):
|
||||
return 0
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
if can_dispatch is not None and not can_dispatch():
|
||||
logger.debug("Cron dispatch paused while gateway drains existing work")
|
||||
return 0
|
||||
|
|
|
|||
|
|
@ -57,6 +57,22 @@ def _resolve_auto_decompose_settings(
|
|||
return enabled, per_tick
|
||||
|
||||
|
||||
def _kanban_dispatch_allowed() -> bool:
|
||||
"""Return False while the global emergency stop (`hermes pause`) is engaged.
|
||||
|
||||
Checked every dispatcher tick BEFORE spawning new workers so a pause takes
|
||||
effect on the next tick without a gateway restart. In-flight workers are
|
||||
never touched — this only stops NEW spawns. Fails open: if the estop
|
||||
module is unimportable, dispatch proceeds (the sentinel gate must not
|
||||
become a new crash surface for the dispatcher).
|
||||
"""
|
||||
try:
|
||||
from agent.estop import check_paused
|
||||
except ImportError:
|
||||
return True
|
||||
return not check_paused("kanban", logger)
|
||||
|
||||
|
||||
def _acquire_singleton_lock(lock_path) -> "tuple[Optional[object], str]":
|
||||
"""Take an exclusive, non-blocking advisory lock for the sole dispatcher.
|
||||
|
||||
|
|
@ -1435,36 +1451,43 @@ class GatewayKanbanWatchersMixin:
|
|||
logger.exception("kanban dispatcher: zombie reaper failed")
|
||||
|
||||
try:
|
||||
# Re-read the auto-decompose toggle live each tick so a user
|
||||
# flipping kanban.auto_decompose=false to STOP runaway fan-out
|
||||
# takes effect on the next tick, not on gateway restart (#49638).
|
||||
_ad_enabled, _ad_per_tick = _read_auto_decompose_settings()
|
||||
if _ad_enabled:
|
||||
await asyncio.to_thread(_auto_decompose_tick, _ad_per_tick)
|
||||
results = await asyncio.to_thread(_tick_once)
|
||||
any_spawned = False
|
||||
for slug, res in (results or []):
|
||||
if res is not None and getattr(res, "spawned", None):
|
||||
any_spawned = True
|
||||
# Quiet by default — only log when something actually
|
||||
# happened, so an idle gateway stays silent.
|
||||
logger.info(
|
||||
"kanban dispatcher [%s]: spawned=%d reclaimed=%d "
|
||||
"crashed=%d timed_out=%d promoted=%d auto_blocked=%d",
|
||||
slug,
|
||||
len(res.spawned),
|
||||
res.reclaimed,
|
||||
len(res.crashed) if hasattr(res.crashed, "__len__") else 0,
|
||||
len(res.timed_out) if hasattr(res.timed_out, "__len__") else 0,
|
||||
res.promoted,
|
||||
len(res.auto_blocked) if hasattr(res.auto_blocked, "__len__") else 0,
|
||||
)
|
||||
# Health telemetry (aggregate across boards)
|
||||
ready_pending = await asyncio.to_thread(_ready_nonempty)
|
||||
if ready_pending and not any_spawned:
|
||||
bad_ticks += 1
|
||||
else:
|
||||
# Global emergency stop (`hermes pause`): skip auto-decompose
|
||||
# and dispatch entirely — no new workers while paused. Running
|
||||
# workers finish naturally; zombie reaping above still runs.
|
||||
if not _kanban_dispatch_allowed():
|
||||
ready_pending = False
|
||||
bad_ticks = 0
|
||||
else:
|
||||
# Re-read the auto-decompose toggle live each tick so a user
|
||||
# flipping kanban.auto_decompose=false to STOP runaway fan-out
|
||||
# takes effect on the next tick, not on gateway restart (#49638).
|
||||
_ad_enabled, _ad_per_tick = _read_auto_decompose_settings()
|
||||
if _ad_enabled:
|
||||
await asyncio.to_thread(_auto_decompose_tick, _ad_per_tick)
|
||||
results = await asyncio.to_thread(_tick_once)
|
||||
any_spawned = False
|
||||
for slug, res in (results or []):
|
||||
if res is not None and getattr(res, "spawned", None):
|
||||
any_spawned = True
|
||||
# Quiet by default — only log when something actually
|
||||
# happened, so an idle gateway stays silent.
|
||||
logger.info(
|
||||
"kanban dispatcher [%s]: spawned=%d reclaimed=%d "
|
||||
"crashed=%d timed_out=%d promoted=%d auto_blocked=%d",
|
||||
slug,
|
||||
len(res.spawned),
|
||||
res.reclaimed,
|
||||
len(res.crashed) if hasattr(res.crashed, "__len__") else 0,
|
||||
len(res.timed_out) if hasattr(res.timed_out, "__len__") else 0,
|
||||
res.promoted,
|
||||
len(res.auto_blocked) if hasattr(res.auto_blocked, "__len__") else 0,
|
||||
)
|
||||
# Health telemetry (aggregate across boards)
|
||||
ready_pending = await asyncio.to_thread(_ready_nonempty)
|
||||
if ready_pending and not any_spawned:
|
||||
bad_ticks += 1
|
||||
else:
|
||||
bad_ticks = 0
|
||||
if bad_ticks >= HEALTH_WINDOW:
|
||||
now = int(time.time())
|
||||
if now - last_warn_at >= 300:
|
||||
|
|
|
|||
|
|
@ -14663,7 +14663,27 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# Record rate limit so subsequent messages are silently ignored
|
||||
pairing_store._record_rate_limit(platform_name, source.user_id)
|
||||
return None
|
||||
|
||||
|
||||
# Global emergency stop (`hermes pause`): give new turns a brief
|
||||
# paused notice instead of starting an agent run. Internal events
|
||||
# (background-process completions from IN-FLIGHT work) bypass the
|
||||
# gate — pause stops NEW work, it never kills or orphans running
|
||||
# work. Placed after auth so unauthorized senders keep the normal
|
||||
# silent/pairing behavior and can't probe pause state.
|
||||
if not is_internal:
|
||||
try:
|
||||
from agent.estop import paused_reply as _estop_paused_reply
|
||||
_paused_notice = _estop_paused_reply()
|
||||
except ImportError:
|
||||
_paused_notice = None
|
||||
if _paused_notice is not None:
|
||||
logger.info(
|
||||
"Gateway turn paused by global emergency stop (platform=%s chat=%s)",
|
||||
getattr(getattr(source, "platform", None), "value", "unknown"),
|
||||
getattr(source, "chat_id", None) or "unknown",
|
||||
)
|
||||
return _paused_notice
|
||||
|
||||
# Intercept messages that are responses to a pending /update prompt.
|
||||
# The update process (detached) wrote .update_prompt.json; the watcher
|
||||
# forwarded it to the user; now the user's reply goes back via
|
||||
|
|
|
|||
|
|
@ -451,6 +451,7 @@ from hermes_cli.subcommands.login import build_login_parser
|
|||
from hermes_cli.subcommands.logout import build_logout_parser
|
||||
from hermes_cli.subcommands.auth import build_auth_parser
|
||||
from hermes_cli.subcommands.status import build_status_parser
|
||||
from hermes_cli.subcommands.pause import build_pause_parser
|
||||
from hermes_cli.subcommands.webhook import build_webhook_parser
|
||||
from hermes_cli.subcommands.hooks import build_hooks_parser
|
||||
from hermes_cli.subcommands.doctor import build_doctor_parser
|
||||
|
|
@ -10609,9 +10610,10 @@ _BUILTIN_SUBCOMMANDS = frozenset(
|
|||
"dump", "egress", "fallback", "gateway", "hooks", "import", "import-agent", "insights",
|
||||
"gui", "desktop", "kanban", "login", "logout", "logs", "lsp", "mcp", "memory", "migrate", "moa",
|
||||
"journey", "memory-graph", "learning",
|
||||
"model", "monitoring", "pairing", "pets", "plugins", "portal", "profile",
|
||||
"model", "monitoring", "pairing", "pause", "pets", "plugins", "portal", "profile",
|
||||
"project", "proxy",
|
||||
"prompt-size",
|
||||
"resume",
|
||||
"send", "sessions", "setup",
|
||||
"skin", "skills", "slack", "status", "sync", "tools", "uninstall", "update",
|
||||
"version", "webhook", "whatsapp", "whatsapp-cloud", "chat", "secrets", "security",
|
||||
|
|
@ -11480,6 +11482,11 @@ def main():
|
|||
# =========================================================================
|
||||
build_status_parser(subparsers, cmd_status=cmd_status)
|
||||
|
||||
# =========================================================================
|
||||
# pause / resume commands (parser built in hermes_cli/subcommands/pause.py)
|
||||
# =========================================================================
|
||||
build_pause_parser(subparsers)
|
||||
|
||||
# =========================================================================
|
||||
# cron command (parser built in hermes_cli/subcommands/cron.py)
|
||||
# =========================================================================
|
||||
|
|
|
|||
|
|
@ -112,6 +112,23 @@ def _effective_provider_label() -> str:
|
|||
from hermes_constants import is_termux as _is_termux
|
||||
|
||||
|
||||
def _estop_status_line():
|
||||
"""One-line pause banner for `hermes status`, or None when not paused.
|
||||
|
||||
Cheap: a single stat on $HERMES_HOME/ESTOP via agent.estop.
|
||||
"""
|
||||
try:
|
||||
from agent.estop import get_state
|
||||
except ImportError:
|
||||
return None
|
||||
state = get_state()
|
||||
if state is None:
|
||||
return None
|
||||
reason = state.get("reason")
|
||||
suffix = f" — reason: {reason}" if reason else ""
|
||||
return f"⏸️ PAUSED (global emergency stop{suffix}; `hermes resume` to lift)"
|
||||
|
||||
|
||||
def show_status(args):
|
||||
"""Show status of all Hermes Agent components."""
|
||||
deep = getattr(args, 'deep', False)
|
||||
|
|
@ -121,6 +138,11 @@ def show_status(args):
|
|||
print(color("│ ⚕ Hermes Agent Status │", Colors.CYAN))
|
||||
print(color("└─────────────────────────────────────────────────────────┘", Colors.CYAN))
|
||||
|
||||
_paused_line = _estop_status_line()
|
||||
if _paused_line:
|
||||
print()
|
||||
print(color(_paused_line, Colors.YELLOW, Colors.BOLD))
|
||||
|
||||
# =========================================================================
|
||||
# Environment
|
||||
# =========================================================================
|
||||
|
|
|
|||
|
|
@ -0,0 +1,70 @@
|
|||
"""``hermes pause`` / ``hermes resume`` — the global emergency stop.
|
||||
|
||||
``hermes pause`` writes the ESTOP sentinel at ``$HERMES_HOME/ESTOP``, which
|
||||
halts cron dispatch, kanban dispatch, and new gateway turns on their next
|
||||
check. In-flight work is never killed. ``hermes resume`` removes the
|
||||
sentinel and normal operation resumes on the next tick — no restart needed.
|
||||
|
||||
Ported from: gastownhall/gastown estop.go (MIT); related prior art:
|
||||
#26778 (/panic — kill/exit semantics, different), #44617.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
|
||||
|
||||
def cmd_pause(args: argparse.Namespace) -> int:
|
||||
"""Engage the global emergency stop."""
|
||||
from agent.estop import engage, get_state, is_engaged
|
||||
|
||||
reason = getattr(args, "reason", None)
|
||||
already = is_engaged()
|
||||
path = engage(reason=reason)
|
||||
state = get_state() or {}
|
||||
verb = "Still paused" if already else "Hermes paused"
|
||||
detail = f" — reason: {state['reason']}" if state.get("reason") else ""
|
||||
print(f"⏸️ {verb}{detail}")
|
||||
print(f" sentinel: {path}")
|
||||
print(
|
||||
" Cron dispatch, kanban dispatch, and new gateway turns are on hold.\n"
|
||||
" In-flight work keeps running. Run `hermes resume` to lift the pause."
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_resume(args: argparse.Namespace) -> int:
|
||||
"""Disengage the global emergency stop."""
|
||||
from agent.estop import disengage, sentinel_path
|
||||
|
||||
if disengage():
|
||||
print("▶️ Hermes resumed — dispatch picks up on the next tick.")
|
||||
else:
|
||||
print(f"Hermes is not paused (no sentinel at {sentinel_path()}).")
|
||||
return 0
|
||||
|
||||
|
||||
def build_pause_parser(subparsers) -> None:
|
||||
"""Attach the ``pause`` and ``resume`` subcommands to ``subparsers``."""
|
||||
pause_parser = subparsers.add_parser(
|
||||
"pause",
|
||||
help="Emergency stop: pause cron/kanban dispatch and new gateway turns",
|
||||
description=(
|
||||
"Engage the global emergency stop. Halts NEW work only — cron "
|
||||
"dispatch, kanban dispatch, and new gateway turns — until "
|
||||
"`hermes resume`. In-flight work is never killed."
|
||||
),
|
||||
)
|
||||
pause_parser.add_argument(
|
||||
"--reason",
|
||||
default=None,
|
||||
help="Optional reason stored in the sentinel and shown to users",
|
||||
)
|
||||
pause_parser.set_defaults(func=cmd_pause)
|
||||
|
||||
resume_parser = subparsers.add_parser(
|
||||
"resume",
|
||||
help="Lift the emergency stop set by `hermes pause`",
|
||||
description="Remove the ESTOP sentinel; dispatch resumes on the next tick.",
|
||||
)
|
||||
resume_parser.set_defaults(func=cmd_resume)
|
||||
|
|
@ -0,0 +1,287 @@
|
|||
"""Global emergency stop (`hermes pause` / `hermes resume`) — agent/estop.py.
|
||||
|
||||
The ESTOP sentinel is a resumable pause for NEW work only: cron dispatch,
|
||||
kanban dispatch, and new gateway turns are halted while it is engaged; work
|
||||
already in flight is never touched. Removing the sentinel (`hermes resume`)
|
||||
restores normal operation with no restart.
|
||||
|
||||
Ported from: gastownhall/gastown estop.go (MIT); related prior art: #26778
|
||||
(/panic — kill/exit semantics, deliberately different) and #44617
|
||||
(interrupt in-flight cron — deliberately NOT done here).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import logging
|
||||
|
||||
import pytest
|
||||
|
||||
from agent import estop
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def hermes_home(tmp_path, monkeypatch):
|
||||
"""Point HERMES_HOME at a temp dir and reset estop module log state."""
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
estop._reset_log_state_for_tests()
|
||||
return tmp_path
|
||||
|
||||
|
||||
# ── sentinel create / remove ────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_engage_creates_sentinel_and_is_engaged(hermes_home):
|
||||
assert estop.is_engaged() is False
|
||||
estop.engage()
|
||||
assert (hermes_home / "ESTOP").exists()
|
||||
assert estop.is_engaged() is True
|
||||
|
||||
|
||||
def test_disengage_removes_sentinel(hermes_home):
|
||||
estop.engage()
|
||||
assert estop.disengage() is True
|
||||
assert not (hermes_home / "ESTOP").exists()
|
||||
assert estop.is_engaged() is False
|
||||
# Disengaging when not engaged is a no-op that reports False.
|
||||
assert estop.disengage() is False
|
||||
|
||||
|
||||
def test_reason_and_timestamp_stored(hermes_home):
|
||||
estop.engage(reason="runaway cron fan-out")
|
||||
state = estop.get_state()
|
||||
assert state is not None
|
||||
assert state["reason"] == "runaway cron fan-out"
|
||||
assert state["engaged_at"] # ISO timestamp string
|
||||
|
||||
raw = json.loads((hermes_home / "ESTOP").read_text(encoding="utf-8"))
|
||||
assert raw["reason"] == "runaway cron fan-out"
|
||||
|
||||
|
||||
def test_get_state_none_when_disengaged(hermes_home):
|
||||
assert estop.get_state() is None
|
||||
|
||||
|
||||
def test_corrupt_sentinel_still_engages(hermes_home):
|
||||
"""A hand-touched/corrupt ESTOP file must still pause (fail safe)."""
|
||||
(hermes_home / "ESTOP").write_text("not json", encoding="utf-8")
|
||||
assert estop.is_engaged() is True
|
||||
state = estop.get_state()
|
||||
assert state is not None
|
||||
assert state.get("reason") is None
|
||||
|
||||
|
||||
# ── paused notice for new gateway turns ─────────────────────────────────────
|
||||
|
||||
|
||||
def test_paused_reply_none_when_disengaged(hermes_home):
|
||||
assert estop.paused_reply() is None
|
||||
|
||||
|
||||
def test_paused_reply_surfaces_reason_and_resume_hint(hermes_home):
|
||||
estop.engage(reason="deploy window")
|
||||
notice = estop.paused_reply()
|
||||
assert notice is not None
|
||||
assert "paused" in notice.lower()
|
||||
assert "deploy window" in notice
|
||||
assert "hermes resume" in notice
|
||||
|
||||
|
||||
def test_paused_reply_without_reason(hermes_home):
|
||||
estop.engage()
|
||||
notice = estop.paused_reply()
|
||||
assert notice is not None
|
||||
assert "paused" in notice.lower()
|
||||
assert "hermes resume" in notice
|
||||
|
||||
|
||||
# ── check_paused: cheap gate + log-once ─────────────────────────────────────
|
||||
|
||||
|
||||
def test_check_paused_logs_once_per_engagement(hermes_home, caplog):
|
||||
logger = logging.getLogger("test.estop.component")
|
||||
estop.engage()
|
||||
with caplog.at_level(logging.INFO, logger=logger.name):
|
||||
assert estop.check_paused("cron", logger) is True
|
||||
assert estop.check_paused("cron", logger) is True
|
||||
assert estop.check_paused("cron", logger) is True
|
||||
paused_logs = [r for r in caplog.records if "paused" in r.getMessage().lower()]
|
||||
assert len(paused_logs) == 1
|
||||
|
||||
# Resume then re-engage → logs once more (transition-based, not forever).
|
||||
caplog.clear()
|
||||
estop.disengage()
|
||||
with caplog.at_level(logging.INFO, logger=logger.name):
|
||||
assert estop.check_paused("cron", logger) is False
|
||||
estop.engage()
|
||||
assert estop.check_paused("cron", logger) is True
|
||||
assert estop.check_paused("cron", logger) is True
|
||||
paused_logs = [r for r in caplog.records if "paused" in r.getMessage().lower()]
|
||||
assert len(paused_logs) == 1
|
||||
|
||||
|
||||
# ── cron scheduler integration ──────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_cron_tick_skips_dispatch_when_engaged(hermes_home, monkeypatch):
|
||||
from cron import scheduler
|
||||
|
||||
calls = []
|
||||
|
||||
def _fake_get_due_jobs():
|
||||
calls.append(1)
|
||||
return []
|
||||
|
||||
monkeypatch.setattr(scheduler, "get_due_jobs", _fake_get_due_jobs)
|
||||
|
||||
estop.engage(reason="test")
|
||||
assert scheduler.tick(verbose=False) == 0
|
||||
assert calls == [], "engaged ESTOP must skip the due-job scan entirely"
|
||||
|
||||
|
||||
def test_cron_tick_resumes_after_disengage(hermes_home, monkeypatch):
|
||||
from cron import scheduler
|
||||
|
||||
calls = []
|
||||
|
||||
def _fake_get_due_jobs():
|
||||
calls.append(1)
|
||||
return []
|
||||
|
||||
monkeypatch.setattr(scheduler, "get_due_jobs", _fake_get_due_jobs)
|
||||
|
||||
estop.engage()
|
||||
scheduler.tick(verbose=False)
|
||||
assert calls == []
|
||||
|
||||
estop.disengage()
|
||||
scheduler.tick(verbose=False)
|
||||
assert calls == [1], "resume must restore normal cron dispatch"
|
||||
|
||||
|
||||
# ── kanban dispatcher integration ───────────────────────────────────────────
|
||||
|
||||
|
||||
def test_kanban_dispatch_blocked_when_engaged(hermes_home):
|
||||
from gateway.kanban_watchers import _kanban_dispatch_allowed
|
||||
|
||||
assert _kanban_dispatch_allowed() is True
|
||||
estop.engage(reason="test")
|
||||
assert _kanban_dispatch_allowed() is False
|
||||
estop.disengage()
|
||||
assert _kanban_dispatch_allowed() is True
|
||||
|
||||
|
||||
# ── gateway turn-start integration ──────────────────────────────────────────
|
||||
|
||||
|
||||
class _FakeSource:
|
||||
platform = None
|
||||
chat_id = "c1"
|
||||
user_id = "u1"
|
||||
user_name = "user"
|
||||
chat_type = "dm"
|
||||
profile = None
|
||||
|
||||
|
||||
class _FakeEvent:
|
||||
internal = False
|
||||
text = "hello"
|
||||
|
||||
def __init__(self):
|
||||
self.source = _FakeSource()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_gateway_new_turn_gets_paused_reply(hermes_home):
|
||||
from gateway.run import GatewayRunner
|
||||
|
||||
runner = object.__new__(GatewayRunner)
|
||||
runner._is_user_authorized = lambda source: True # bare-instance stub
|
||||
estop.engage(reason="maintenance")
|
||||
reply = await runner._handle_message(_FakeEvent())
|
||||
assert reply is not None
|
||||
assert "paused" in reply.lower()
|
||||
assert "maintenance" in reply
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_gateway_internal_events_bypass_estop(hermes_home):
|
||||
"""Internal events (in-flight work completions) must NOT be paused."""
|
||||
from gateway.run import GatewayRunner
|
||||
|
||||
runner = object.__new__(GatewayRunner)
|
||||
estop.engage()
|
||||
event = _FakeEvent()
|
||||
event.internal = True
|
||||
# An internal event proceeds past the estop gate; the bare runner then
|
||||
# blows up further down the pipeline on missing attributes — that error
|
||||
# (anything but a paused reply) proves the gate let it through.
|
||||
try:
|
||||
reply = await runner._handle_message(event)
|
||||
except Exception:
|
||||
return
|
||||
assert reply is None or "paused" not in (reply or "").lower()
|
||||
|
||||
|
||||
# ── CLI: hermes pause / hermes resume ───────────────────────────────────────
|
||||
|
||||
|
||||
def test_cli_pause_engages_with_reason(hermes_home, capsys):
|
||||
from hermes_cli.subcommands.pause import cmd_pause
|
||||
|
||||
rc = cmd_pause(argparse.Namespace(reason="ops incident"))
|
||||
assert rc == 0
|
||||
assert estop.is_engaged() is True
|
||||
assert estop.get_state()["reason"] == "ops incident"
|
||||
assert "paused" in capsys.readouterr().out.lower()
|
||||
|
||||
|
||||
def test_cli_pause_idempotent(hermes_home, capsys):
|
||||
from hermes_cli.subcommands.pause import cmd_pause
|
||||
|
||||
assert cmd_pause(argparse.Namespace(reason=None)) == 0
|
||||
assert cmd_pause(argparse.Namespace(reason=None)) == 0
|
||||
assert estop.is_engaged() is True
|
||||
|
||||
|
||||
def test_cli_resume_disengages(hermes_home, capsys):
|
||||
from hermes_cli.subcommands.pause import cmd_pause, cmd_resume
|
||||
|
||||
cmd_pause(argparse.Namespace(reason=None))
|
||||
rc = cmd_resume(argparse.Namespace())
|
||||
assert rc == 0
|
||||
assert estop.is_engaged() is False
|
||||
assert "resumed" in capsys.readouterr().out.lower()
|
||||
|
||||
|
||||
def test_cli_resume_when_not_paused(hermes_home, capsys):
|
||||
from hermes_cli.subcommands.pause import cmd_resume
|
||||
|
||||
rc = cmd_resume(argparse.Namespace())
|
||||
assert rc == 0
|
||||
assert "not paused" in capsys.readouterr().out.lower()
|
||||
|
||||
|
||||
def test_builtin_subcommands_include_pause_resume():
|
||||
from hermes_cli.main import _BUILTIN_SUBCOMMANDS
|
||||
|
||||
assert "pause" in _BUILTIN_SUBCOMMANDS
|
||||
assert "resume" in _BUILTIN_SUBCOMMANDS
|
||||
|
||||
|
||||
# ── hermes status surfacing ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_status_line_when_paused(hermes_home):
|
||||
from hermes_cli.status import _estop_status_line
|
||||
|
||||
assert _estop_status_line() is None
|
||||
estop.engage(reason="ops")
|
||||
line = _estop_status_line()
|
||||
assert line is not None
|
||||
assert "paused" in line.lower()
|
||||
assert "ops" in line
|
||||
estop.disengage()
|
||||
assert _estop_status_line() is None
|
||||
Loading…
Reference in New Issue