hermes gateway stop no longer exits 1 (and gets revived) when the stop watcher beats the CLI's SIGTERM

gateway/run.py

@@ -5264,6 +5264,8 @@ def restart_signal_handler():
 
 def _start_gateway_make_shutdown_signal_handler(runner, _signal_initiated_shutdown: list):
     """Build the SIGINT/SIGTERM handler; ``_signal_initiated_shutdown[0]`` records an unplanned signal."""
+    planned_stop_seen = [False]
+
     def shutdown_signal_handler(received_signal=None):
         # Planned --replace takeover (sibling marked this PID): exit 0 so systemd won't revive us.
         def _takeover() -> bool:
@@ -5283,6 +5285,12 @@ def _snapshot():
         planned_takeover = bool(_best_effort(_takeover, "Takeover marker check failed: %s"))
         planned_stop = received_signal == signal.SIGINT or (
             not planned_takeover and bool(_best_effort(_planned_stop, "Planned stop marker check failed: %s")))
+        # `hermes gateway stop` writes the marker, THEN signals: the planned-stop watcher can consume
+        # the marker in between, and the CLI's own SIGTERM must not then read as an external kill.
+        if planned_stop:
+            planned_stop_seen[0] = True
+        elif planned_stop_seen[0] and not planned_takeover:
+            planned_stop = True
         _shutdown_ctx = _best_effort(_snapshot, "snapshot_shutdown_context failed: %s")
         sig_name = _shutdown_ctx["signal"] if _shutdown_ctx else None
 

tests/gateway/test_planned_stop_watcher.py

@@ -201,3 +201,35 @@ def test_planned_stop_marker_targets_self_probe_is_non_destructive(tmp_path, mon
     assert status_mod.planned_stop_marker_targets_self() is True
 
 
+def test_cli_sigterm_after_watcher_consumed_the_marker_stays_planned(tmp_path, monkeypatch):
+    """`hermes gateway stop` writes the marker, THEN signals. When the watcher fires in between and
+    consumes the marker, the trailing SIGTERM must not read as an external kill (exit 1 → a
+    Restart=on-failure supervisor revives the gateway the operator stopped). A bare SIGTERM with no
+    planned stop before it is still an external kill."""
+    import signal
+    from types import SimpleNamespace
+
+    import gateway.run as run_mod
+    import gateway.shutdown_forensics as forensics
+
+    marker = tmp_path / ".gateway-planned-stop.json"
+    monkeypatch.setattr(status_mod, "_get_planned_stop_marker_path", lambda: marker)
+    monkeypatch.setattr(status_mod, "consume_takeover_marker_for_self", lambda: False)
+    monkeypatch.setattr(forensics, "snapshot_shutdown_context", lambda *a, **k: None)
+    monkeypatch.setattr(run_mod.asyncio, "create_task", lambda coro: coro.close())
+
+    def _handler():
+        runner = SimpleNamespace(_signal_initiated_shutdown=False, stop=lambda: asyncio.sleep(0))
+        flag = [False]
+        return run_mod._start_gateway_make_shutdown_signal_handler(runner, flag), flag
+
+    handler, flag = _handler()
+    _write_self_marker(marker)
+    handler(None)  # the watcher's call: consumes the marker
+    assert not marker.exists()
+    handler(signal.SIGTERM)  # the CLI's own SIGTERM, marker already gone
+    assert flag[0] is False
+
+    bare, bare_flag = _handler()
+    bare(signal.SIGTERM)
+    assert bare_flag[0] is True