test(gateway): cover stale gateway_state.json detection (TTL + PID liveness)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

tests/hermes_cli/test_gateway_runtime_health.py

@@ -1,6 +1,119 @@
+from datetime import datetime, timedelta, timezone
+
 from hermes_cli.gateway import _runtime_health_lines
 
 
+def _iso_age(seconds_ago: float) -> str:
+    """ISO-8601 UTC timestamp ``seconds_ago`` in the past (drives _marker_is_stale)."""
+    return (datetime.now(timezone.utc) - timedelta(seconds=seconds_ago)).isoformat()
+
+
+_STALE_LINE_PREFIX = "⚠ Stale gateway_state.json:"
+
+
+def _stale_lines(lines):
+    return [ln for ln in lines if ln.startswith(_STALE_LINE_PREFIX)]
+
+
+def test_runtime_health_lines_flags_stale_running_with_dead_pid(monkeypatch):
+    """Stale updated_at + dead PID + 'running' -> contradiction line, no draining line."""
+    from gateway import status as status_mod
+
+    monkeypatch.setattr(
+        "gateway.status.read_runtime_status",
+        lambda: {
+            "gateway_state": "running",
+            "pid": 4242,
+            "start_time": 111,
+            "updated_at": _iso_age(600),  # well past the 120s TTL -> stale
+            "active_agents": 0,
+        },
+    )
+    # Recorded PID is gone (ungraceful kill); no real process is touched.
+    monkeypatch.setattr(status_mod, "_pid_exists", lambda pid: False)
+    monkeypatch.setattr(status_mod, "_get_process_start_time", lambda pid: None)
+
+    lines = _runtime_health_lines()
+
+    stale = _stale_lines(lines)
+    assert len(stale) == 1, lines
+    assert "recorded state 'running'" in stale[0]
+    assert "recorded process is gone" in stale[0]
+    # The misleading live-state summary must be suppressed.
+    assert not any("draining" in ln.lower() for ln in lines), lines
+
+
+def test_runtime_health_lines_no_warning_for_fresh_running_with_live_pid(monkeypatch):
+    """Fresh updated_at + live PID + 'running' -> no contradiction line."""
+    from gateway import status as status_mod
+
+    monkeypatch.setattr(
+        "gateway.status.read_runtime_status",
+        lambda: {
+            "gateway_state": "running",
+            "pid": 4242,
+            "start_time": 111,
+            "updated_at": _iso_age(5),  # fresh, within the 120s TTL
+            "active_agents": 2,
+        },
+    )
+    monkeypatch.setattr(status_mod, "_pid_exists", lambda pid: True)
+    # start_time matches the record -> not a recycled PID.
+    monkeypatch.setattr(status_mod, "_get_process_start_time", lambda pid: 111)
+
+    lines = _runtime_health_lines()
+
+    assert _stale_lines(lines) == [], lines
+
+
+def test_runtime_health_lines_no_warning_for_stale_running_but_live_pid(monkeypatch):
+    """Stale updated_at but PID still live (long-idle gateway) -> no false positive."""
+    from gateway import status as status_mod
+
+    monkeypatch.setattr(
+        "gateway.status.read_runtime_status",
+        lambda: {
+            "gateway_state": "running",
+            "pid": 4242,
+            "start_time": 111,
+            "updated_at": _iso_age(600),  # stale timestamp...
+            "active_agents": 0,
+        },
+    )
+    # ...but the recorded process is genuinely alive (start_time matches).
+    monkeypatch.setattr(status_mod, "_pid_exists", lambda pid: pid == 4242)
+    monkeypatch.setattr(status_mod, "_get_process_start_time", lambda pid: 111)
+
+    lines = _runtime_health_lines()
+
+    # Guard requires BOTH stale AND dead PID; live PID must suppress the warning.
+    assert _stale_lines(lines) == [], lines
+
+
+def test_runtime_health_lines_treats_missing_updated_at_as_stale(monkeypatch):
+    """Missing updated_at (degrade path) + dead PID + 'running' -> contradiction line."""
+    from gateway import status as status_mod
+
+    monkeypatch.setattr(
+        "gateway.status.read_runtime_status",
+        lambda: {
+            "gateway_state": "running",
+            "pid": 4242,
+            "start_time": 111,
+            # no "updated_at" -> runtime_status_is_stale must treat as stale
+            "active_agents": 0,
+        },
+    )
+    monkeypatch.setattr(status_mod, "_pid_exists", lambda pid: False)
+    monkeypatch.setattr(status_mod, "_get_process_start_time", lambda pid: None)
+
+    lines = _runtime_health_lines()
+
+    stale = _stale_lines(lines)
+    assert len(stale) == 1, lines
+    assert "recorded state 'running'" in stale[0]
+
+
 def test_runtime_health_lines_include_fatal_platform_and_startup_reason(monkeypatch):
     monkeypatch.setattr(
         "gateway.status.read_runtime_status",