feat(gateway): structured start-blocked telemetry + crash-loop visibility on /api/status

Implements the trace-point suggestions from the Hermes Cloud health report.

A live gateway plus a service supervisor (systemd Restart=, s6, Fly machine
restart) respawning a second 'gateway run' produces a tight crash-loop: every
losing starter refuses to claim the PID file / runtime lock and exits, while
the original gateway stays up. A point-in-time /api/status probe reports the
box as healthy (gateway_running: true) so the loop is invisible. Confirmed in
prod on girt-numbat-4665: probe healthy while emitting ~24k 'Gateway already
running' lines/24h, only detectable via a fragile ILIKE log scan.

- gateway/status.py: add record_start_blocked(reason, pid) and
  count_recent_start_blocks(window). record_start_blocked emits a single
  canonical, exact-match log token 'gateway.start_blocked' with stable
  reason=/pid=/version= key=value fields (replacing the ILIKE '%already
  running%' scan and making loops attributable to a release cohort), and
  persists a self-pruning, capped ring of recent block timestamps under
  HERMES_HOME so a separate process can observe the loop.
- gateway/run.py: instrument all four start-refusal sites (already_running,
  runtime_lock_held, pid_file_race, startup_race) with accurate reason tokens.
  Existing log messages are preserved unchanged (tests assert on them).
- hermes_cli/web_server.py + gateway/platforms/api_server.py: surface
  gateway_restarts_5m on /api/status and /health/detailed, read from the
  shared ring, so a wedged-but-looping box stops reading as cleanly healthy.

No outbound telemetry, no new env var, no third-party identifiers: this is
local log + existing public liveness fields only. version is already exposed
on /api/status, so the 'app_version cohort' suggestion is realized as the
version= field on the structured log token rather than a Prometheus label
(the repo ships no Prometheus exporter).

Tests: 9 new in tests/gateway/test_status.py (record/count, windowing,
self-prune, cap, corrupt-file degradation, canonical log token) and 2 in
tests/hermes_cli/test_web_server.py (field present + reflects the ring).
0 new failures in the gateway suite.

gateway/platforms/api_server.py

@@ -1147,6 +1147,7 @@ class APIServerAdapter(BasePlatformAdapter):
         /proc access.  No authentication required.
         """
         from gateway.status import (
+            count_recent_start_blocks,
             derive_gateway_busy,
             derive_gateway_drainable,
             parse_active_agents,
@@ -1175,6 +1176,10 @@ class APIServerAdapter(BasePlatformAdapter):
                 gateway_running=True,
                 gateway_state=gw_state,
             ),
+            # Mirror /api/status: refused-start crash-loop count over 5 min, read
+            # from the shared HERMES_HOME ring so a cross-container dashboard
+            # probe sees the same wedged-but-looping signal /api/status exposes.
+            "gateway_restarts_5m": count_recent_start_blocks(300),
             "exit_reason": runtime.get("exit_reason"),
             "updated_at": runtime.get("updated_at"),
             "pid": os.getpid(),

gateway/run.py

@@ -17385,6 +17385,7 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool =
         acquire_gateway_runtime_lock,
         get_running_pid,
         get_process_start_time,
+        record_start_blocked,
         release_gateway_runtime_lock,
         remove_pid_file,
         terminate_pid,
@@ -17499,6 +17500,7 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool =
                 pass
         else:
             hermes_home = str(get_hermes_home())
+            record_start_blocked("already_running", pid=existing_pid)
             logger.error(
                 "Another gateway instance is already running (PID %d, HERMES_HOME=%s). "
                 "Use 'hermes gateway restart' to replace it, or 'hermes gateway stop' first.",
@@ -17725,15 +17727,17 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool =
     # Telegram polling, Discord gateway sockets, etc. The loser exits
     # cleanly before touching any external service.
     import atexit
-    from gateway.status import write_pid_file, remove_pid_file, get_running_pid
+    from gateway.status import write_pid_file, remove_pid_file, get_running_pid, record_start_blocked
     _current_pid = get_running_pid()
     if _current_pid is not None and _current_pid != os.getpid():
+        record_start_blocked("startup_race", pid=_current_pid)
         logger.error(
             "Another gateway instance (PID %d) started during our startup. "
             "Exiting to avoid double-running.", _current_pid
         )
         return False
     if not acquire_gateway_runtime_lock():
+        record_start_blocked("runtime_lock_held")
         logger.error(
             "Gateway runtime lock is already held by another instance. Exiting."
         )
@@ -17742,6 +17746,7 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool =
         write_pid_file()
     except FileExistsError:
         release_gateway_runtime_lock()
+        record_start_blocked("pid_file_race")
         logger.error(
             "PID file race lost to another gateway instance. Exiting."
         )

gateway/status.py

@@ -646,6 +646,171 @@ def read_runtime_status() -> Optional[dict[str, Any]]:
     return _read_json_file(_get_runtime_status_path())
 
 
+# ---------------------------------------------------------------------------
+# Start-blocked accounting (crash-loop observability)
+#
+# When a gateway start is refused because another instance already holds the
+# PID file / runtime lock, the losing process logs an error and exits.  Under a
+# service supervisor (systemd Restart=, s6, Fly machine restart) that refused
+# starter is respawned immediately, producing a tight crash-loop: a box whose
+# *live* gateway is fine but is being hammered by a second would-be gateway
+# that can never win.  A point-in-time ``/api/status`` probe reports such a box
+# as healthy (``gateway_running: true``) because the original gateway IS up —
+# the loop is invisible to the probe.  Confirmed in production 2026-06 on
+# ``girt-numbat-4665``: probe healthy while emitting ~24k "Gateway already
+# running" lines/24h.
+#
+# Two observability gaps are closed here:
+#   1. The crash-loop was only detectable by an ``ILIKE '%already running%'``
+#      scan over free-text log messages — fragile (message wording can drift)
+#      and expensive (full-text scan).  ``record_start_blocked`` emits a single
+#      canonical, exact-match token ``gateway.start_blocked`` plus stable
+#      ``reason=`` / ``pid=`` / ``version=`` key=value pairs, so fleet telemetry
+#      can count loops with an exact substring match and attribute them to a
+#      version cohort.
+#   2. The probe couldn't see the loop at all.  ``record_start_blocked``
+#      persists a small ring of recent block timestamps to a file under
+#      HERMES_HOME; the dashboard ``/api/status`` handler — a SEPARATE process
+#      sharing the same HERMES_HOME — reads it via ``count_recent_start_blocks``
+#      and surfaces a rolling count, so a wedged-but-looping box stops reading
+#      as cleanly healthy.
+# ---------------------------------------------------------------------------
+
+_START_BLOCKS_FILE = "gateway_start_blocks.json"
+# Cap the persisted ring so a sustained loop can't grow the file unbounded.
+_START_BLOCKS_MAX_ENTRIES = 128
+# Drop entries older than this when recording, so the file self-prunes without
+# a separate sweeper.  An hour comfortably covers the 5-minute window the probe
+# reports while still bounding growth.
+_START_BLOCKS_RETENTION_S = 3600
+# Default window the probe reports on.
+START_BLOCKS_DEFAULT_WINDOW_S = 300
+
+# Stable reason tokens.  Telemetry matches on these exact strings, so treat the
+# set as a contract — add new tokens, don't reword existing ones.
+START_BLOCK_REASONS = frozenset(
+    {
+        "already_running",  # a live gateway holds the PID file before we start
+        "runtime_lock_held",  # could not acquire the cross-process runtime lock
+        "pid_file_race",  # lost the O_CREAT|O_EXCL PID-file race
+        "startup_race",  # another gateway appeared during our async startup
+    }
+)
+
+
+def _get_start_blocks_path() -> Path:
+    """Return the persisted start-blocked ring file path (sibling of PID file)."""
+    return _get_pid_path().with_name(_START_BLOCKS_FILE)
+
+
+def record_start_blocked(reason: str, *, pid: Optional[int] = None) -> int:
+    """Record a refused gateway start and emit a canonical structured log line.
+
+    ``reason`` should be one of :data:`START_BLOCK_REASONS`; unknown values are
+    still recorded (logged with ``reason=other:<value>``) so a miswired caller
+    is visible rather than silently dropped.
+
+    ``pid`` is the PID of the gateway that won (the live instance we collided
+    with), when known — useful for confirming the loop is colliding with a
+    single stable winner.
+
+    Persists a timestamped entry to a small ring file under HERMES_HOME so the
+    dashboard ``/api/status`` handler (a separate process) can observe the loop,
+    and emits one ``gateway.start_blocked`` log token for exact-match telemetry.
+
+    Returns the number of start-blocks recorded within the retention window
+    (including this one), or ``0`` if persistence failed.
+    """
+    canonical_reason = reason if reason in START_BLOCK_REASONS else f"other:{reason}"
+    now = datetime.now(timezone.utc).timestamp()
+
+    recorded_count = 0
+    path = _get_start_blocks_path()
+    try:
+        payload = _read_json_file(path) or {}
+        entries = payload.get("entries")
+        if not isinstance(entries, list):
+            entries = []
+        # Self-prune: drop anything outside the retention window, then cap.
+        cutoff = now - _START_BLOCKS_RETENTION_S
+        entries = [
+            e
+            for e in entries
+            if isinstance(e, dict)
+            and isinstance(e.get("ts"), (int, float))
+            and e["ts"] >= cutoff
+        ]
+        entry: dict[str, Any] = {"ts": now, "reason": canonical_reason}
+        if pid is not None:
+            entry["pid"] = pid
+        entries.append(entry)
+        if len(entries) > _START_BLOCKS_MAX_ENTRIES:
+            entries = entries[-_START_BLOCKS_MAX_ENTRIES:]
+        _write_json_file(path, {"entries": entries})
+        recorded_count = len(entries)
+    except Exception:
+        # Persistence is best-effort observability — never let it break startup.
+        recorded_count = 0
+
+    # Single canonical, exact-match-greppable token.  Hermes logs are plain-text
+    # (RotatingFileHandler), so the "structured field" is realized as stable
+    # key=value pairs inside the message — ``gateway.start_blocked`` is the
+    # exact substring telemetry keys off, replacing the fragile
+    # ``ILIKE '%already running%'`` scan, and ``version=`` makes crash-loops
+    # attributable to a release cohort.
+    try:
+        import logging as _logging
+
+        from hermes_cli import __version__ as _hermes_version
+    except Exception:
+        _hermes_version = "unknown"
+        import logging as _logging  # type: ignore[no-redef]
+
+    pid_field = pid if pid is not None else ""
+    _logging.getLogger("gateway.status").error(
+        "gateway.start_blocked reason=%s pid=%s version=%s",
+        canonical_reason,
+        pid_field,
+        _hermes_version,
+    )
+    return recorded_count
+
+
+def count_recent_start_blocks(
+    window_seconds: int = START_BLOCKS_DEFAULT_WINDOW_S,
+) -> int:
+    """Return the number of refused gateway starts recorded within ``window_seconds``.
+
+    Reads the persisted ring written by :func:`record_start_blocked`.  Returns
+    ``0`` when the file is absent, unreadable, or holds no recent entries — a
+    healthy single-gateway box never writes it, so the common case is a cheap
+    missing-file stat.  Safe for the public ``/api/status`` liveness probe.
+    """
+    try:
+        window = int(window_seconds)
+    except (TypeError, ValueError):
+        window = START_BLOCKS_DEFAULT_WINDOW_S
+    if window <= 0:
+        return 0
+
+    payload = _read_json_file(_get_start_blocks_path())
+    if not isinstance(payload, dict):
+        return 0
+    entries = payload.get("entries")
+    if not isinstance(entries, list):
+        return 0
+
+    cutoff = datetime.now(timezone.utc).timestamp() - window
+    count = 0
+    for e in entries:
+        if not isinstance(e, dict):
+            continue
+        ts = e.get("ts")
+        if isinstance(ts, (int, float)) and ts >= cutoff:
+            count += 1
+    return count
+
+
 def parse_active_agents(raw: Any) -> int:
     """Coerce a persisted ``active_agents`` value to a clamped non-negative int.
 

hermes_cli/web_server.py

@@ -70,6 +70,7 @@ from hermes_cli.memory_providers import (
     get_memory_provider,
 )
 from gateway.status import (
+    count_recent_start_blocks,
     derive_gateway_busy,
     derive_gateway_drainable,
     get_running_pid,
@@ -1940,6 +1941,16 @@ async def get_status(profile: Optional[str] = None):
             # Module not importable yet (early startup) — leave as [].
             pass
 
+        # Refused-start (crash-loop) counter over the last 5 minutes.  A live
+        # gateway plus a supervisor that keeps respawning a second `gateway
+        # run` produces a tight loop where each loser refuses to start and
+        # exits; the original gateway stays up, so ``gateway_running`` reads
+        # healthy and the loop is otherwise invisible to this probe.  Surfacing
+        # the count here lets uptime/fleet monitors flag a wedged-but-looping
+        # box (confirmed prod case: probe healthy while looping ~24k times/24h).
+        # Cheap missing-file stat on a healthy single-gateway box.
+        gateway_restarts_5m = count_recent_start_blocks(300)
+
         # Always-public liveness + auth-gate shape. Safe for external uptime
         # probes (NAS's wildcard-subdomain liveness probe), the SPA's pre-login
         # bootstrap, and anyone who can curl the host — i.e. exactly the audience
@@ -1955,6 +1966,7 @@ async def get_status(profile: Optional[str] = None):
             "gateway_platforms": gateway_platforms,
             "gateway_exit_reason": gateway_exit_reason,
             "gateway_updated_at": gateway_updated_at,
+            "gateway_restarts_5m": gateway_restarts_5m,
             "active_agents": active_agents,
             "gateway_busy": gateway_busy,
             "gateway_drainable": gateway_drainable,

tests/gateway/test_status.py

@@ -1254,3 +1254,112 @@ class TestGatewayBusyDerivation:
             assert status.derive_gateway_drainable(
                 gateway_running=True, gateway_state=state
             ) is False, state
+
+
+class TestStartBlockedAccounting:
+    """Crash-loop observability: record_start_blocked + count_recent_start_blocks.
+
+    A refused gateway start (another instance already holds the PID file / lock)
+    is recorded to a small ring under HERMES_HOME and emits a canonical
+    ``gateway.start_blocked`` log token. The dashboard ``/api/status`` handler —
+    a separate process sharing HERMES_HOME — reads the ring to surface a rolling
+    count, so a wedged-but-looping box (live gateway + a supervisor respawning a
+    losing starter) stops reading as cleanly healthy.
+    """
+
+    def test_healthy_box_has_no_file_and_counts_zero(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        # No start ever blocked: the file must not exist and count is 0.
+        assert not (tmp_path / "gateway_start_blocks.json").exists()
+        assert status.count_recent_start_blocks() == 0
+        assert status.count_recent_start_blocks(300) == 0
+
+    def test_record_creates_ring_and_count_reflects_it(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+
+        n = status.record_start_blocked("already_running", pid=4242)
+        assert n == 1
+        ring = tmp_path / "gateway_start_blocks.json"
+        assert ring.exists()
+        payload = json.loads(ring.read_text())
+        assert isinstance(payload["entries"], list)
+        assert payload["entries"][0]["reason"] == "already_running"
+        assert payload["entries"][0]["pid"] == 4242
+        assert isinstance(payload["entries"][0]["ts"], (int, float))
+
+        # A second block accumulates and the rolling count tracks it.
+        assert status.record_start_blocked("runtime_lock_held") == 2
+        assert status.count_recent_start_blocks(300) == 2
+
+    def test_unknown_reason_is_recorded_namespaced_not_dropped(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        status.record_start_blocked("totally_made_up")
+        payload = json.loads((tmp_path / "gateway_start_blocks.json").read_text())
+        # A miswired caller is visible (namespaced), never silently dropped.
+        assert payload["entries"][0]["reason"] == "other:totally_made_up"
+        assert status.count_recent_start_blocks() == 1
+
+    def test_count_excludes_entries_outside_window(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        import time as _time
+
+        now = _time.time()
+        ring = tmp_path / "gateway_start_blocks.json"
+        # Hand-write a ring: one recent (60s ago), one old (600s ago).
+        ring.write_text(json.dumps({"entries": [
+            {"ts": now - 600, "reason": "already_running"},
+            {"ts": now - 60, "reason": "already_running"},
+        ]}))
+        # 5-minute window sees only the recent one.
+        assert status.count_recent_start_blocks(300) == 1
+        # A wider 20-minute window sees both.
+        assert status.count_recent_start_blocks(1200) == 2
+
+    def test_record_self_prunes_beyond_retention(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        import time as _time
+
+        now = _time.time()
+        ring = tmp_path / "gateway_start_blocks.json"
+        # Seed an entry older than the 1h retention; recording must drop it.
+        ring.write_text(json.dumps({"entries": [
+            {"ts": now - status._START_BLOCKS_RETENTION_S - 10, "reason": "already_running"},
+        ]}))
+        n = status.record_start_blocked("already_running")
+        # Old entry pruned, only the fresh one remains.
+        assert n == 1
+        payload = json.loads(ring.read_text())
+        assert len(payload["entries"]) == 1
+
+    def test_ring_is_capped(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        for _ in range(status._START_BLOCKS_MAX_ENTRIES + 25):
+            status.record_start_blocked("already_running")
+        payload = json.loads((tmp_path / "gateway_start_blocks.json").read_text())
+        assert len(payload["entries"]) == status._START_BLOCKS_MAX_ENTRIES
+
+    def test_count_degrades_on_corrupt_file(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        ring = tmp_path / "gateway_start_blocks.json"
+        ring.write_text("{ this is not json ]")
+        # Unreadable ring degrades to 0, never raises on the public probe path.
+        assert status.count_recent_start_blocks(300) == 0
+
+    def test_count_clamps_non_positive_and_bad_window(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        status.record_start_blocked("already_running")
+        assert status.count_recent_start_blocks(0) == 0
+        assert status.count_recent_start_blocks(-5) == 0
+        # Non-int window falls back to the default window (entry is recent).
+        assert status.count_recent_start_blocks("garbage") == 1
+
+    def test_emits_canonical_log_token(self, tmp_path, monkeypatch, caplog):
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        import logging
+
+        with caplog.at_level(logging.ERROR, logger="gateway.status"):
+            status.record_start_blocked("already_running", pid=99)
+        # Exact-match token telemetry keys off, plus stable key=value fields.
+        msgs = [r.getMessage() for r in caplog.records]
+        assert any("gateway.start_blocked" in m for m in msgs)
+        assert any("reason=already_running" in m and "pid=99" in m and "version=" in m for m in msgs)

tests/hermes_cli/test_web_server.py

@@ -248,6 +248,22 @@ class TestWebServerEndpoints:
         assert "hermes_home" in data
         assert "active_sessions" in data
         assert data["can_update_hermes"] is True
+        # Crash-loop counter is always present; healthy box reports 0.
+        assert "gateway_restarts_5m" in data
+        assert data["gateway_restarts_5m"] == 0
+
+    def test_get_status_surfaces_start_block_crashloop_count(self):
+        """A live gateway being hammered by refused restarts must surface a
+        non-zero ``gateway_restarts_5m`` even though ``gateway_running`` is True —
+        the blind spot a point-in-time probe otherwise misses."""
+        from gateway import status as gw_status
+
+        gw_status.record_start_blocked("already_running", pid=4242)
+        gw_status.record_start_blocked("already_running", pid=4242)
+
+        resp = self.client.get("/api/status")
+        assert resp.status_code == 200
+        assert resp.json()["gateway_restarts_5m"] == 2
 
     def test_get_status_hides_update_capability_in_managed_runtime(self, monkeypatch):
         import hermes_cli.web_server as web_server