From 5db1b72b1fd3aaaba740d70148373b1c4abf8dcc Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Fri, 7 Aug 2026 08:09:12 -0700 Subject: [PATCH] =?UTF-8?q?feat(cli):=20global=20emergency=20stop=20?= =?UTF-8?q?=E2=80=94=20`hermes=20pause`=20/=20`hermes=20resume`?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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). --- agent/estop.py | 167 +++++++++++++++++++ cron/scheduler.py | 11 ++ gateway/kanban_watchers.py | 81 +++++---- gateway/run.py | 22 ++- hermes_cli/main.py | 9 +- hermes_cli/status.py | 22 +++ hermes_cli/subcommands/pause.py | 70 ++++++++ tests/test_estop.py | 287 ++++++++++++++++++++++++++++++++ 8 files changed, 638 insertions(+), 31 deletions(-) create mode 100644 agent/estop.py create mode 100644 hermes_cli/subcommands/pause.py create mode 100644 tests/test_estop.py diff --git a/agent/estop.py b/agent/estop.py new file mode 100644 index 0000000000000..1fe196bca0cee --- /dev/null +++ b/agent/estop.py @@ -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() diff --git a/cron/scheduler.py b/cron/scheduler.py index c2ad50909d912..8c240cbdd1196 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -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 diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index dc2c5efd9011a..cad31e993e3fa 100644 --- a/gateway/kanban_watchers.py +++ b/gateway/kanban_watchers.py @@ -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: diff --git a/gateway/run.py b/gateway/run.py index 9a9c98cb74cbc..84ada4ee34bbc 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -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 diff --git a/hermes_cli/main.py b/hermes_cli/main.py index e6612636da75e..d71a4f2216cbe 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -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) # ========================================================================= diff --git a/hermes_cli/status.py b/hermes_cli/status.py index 110e0f5696149..47ac61381d634 100644 --- a/hermes_cli/status.py +++ b/hermes_cli/status.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 # ========================================================================= diff --git a/hermes_cli/subcommands/pause.py b/hermes_cli/subcommands/pause.py new file mode 100644 index 0000000000000..048f7839eee7a --- /dev/null +++ b/hermes_cli/subcommands/pause.py @@ -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) diff --git a/tests/test_estop.py b/tests/test_estop.py new file mode 100644 index 0000000000000..820f9aa6c4464 --- /dev/null +++ b/tests/test_estop.py @@ -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