v6.7 Tranche 1: kanban_complete verification gates (#28, #62, #64)

cli.py

@@ -3109,6 +3109,13 @@ def __init__(
         # Initialize Rich console
         self.console = Console()
         self.config = CLI_CONFIG
+        # Stash for the most recent chat() turn's run_conversation result.
+        # Single-query callers in main() inspect this AFTER chat() returns to
+        # decide the worker exit code (e.g. rate-limit → exit 75 per
+        # `_worker_exit_code_from_result`). Initialized here so the
+        # invariant "stash is reset at turn boundary" doesn't depend on
+        # any early returns inside chat().
+        self._last_run_result = None
         self.compact = compact if compact is not None else CLI_CONFIG["display"].get("compact", False)
         # tool_progress: "off", "new", "all", "verbose" (from config.yaml display section)
         # YAML 1.1 parses bare `off` as boolean False — normalise to string.
@@ -12350,6 +12357,12 @@ def chat(self, message, images: list = None) -> Optional[str]:
         ChatConsole().print(f"[{_accent_hex()}]{'─' * 40}[/]")
         print(flush=True)
 
+        # Reset stash at turn boundary so a downstream consumer (e.g. the
+        # non-quiet `chat -q "..."` exit-code mapper) never reads a stale
+        # failure dict from a previous turn. Set explicitly here so the
+        # invariant doesn't depend on the order of early returns below.
+        self._last_run_result = None
+
         try:
             # Run the conversation with interrupt monitoring
             result = None
@@ -12827,10 +12840,26 @@ def run_agent():
                 print(f"\n⏩ Delivering leftover /steer as next turn: '{preview}'")
                 self._pending_input.put(_leftover_steer)
 
+            # Stash the run result so single-query callers can inspect
+            # failure_reason (rate_limit/billing) for kanban-worker exit code
+            # mapping. See `_worker_exit_code_from_result` and hermes-agent#5.
+            # chat() itself returns only the response string for back-compat;
+            # this attribute is the only out-of-band channel for run metadata.
+            self._last_run_result = result
+
             return response
 
         except Exception as e:
             print(f"Error: {e}")
+            # Synthesize a failure result so single-query callers downstream
+            # can still apply the kanban exit-code mapping (a thrown exception
+            # is by definition a failure — `failure_reason` is unknown so the
+            # rate-limit special case won't fire, and we fall through to exit 1).
+            self._last_run_result = {
+                "failed": True,
+                "error": str(e)[:300],
+                "completed": False,
+            }
             return None
         finally:
             # Ensure streaming TTS resources are cleaned up even on error.
@@ -15697,6 +15726,39 @@ def _block(reason: str) -> None:
     )
 
 
+def _worker_exit_code_from_result(result) -> int:
+    """Map a run_conversation/chat result dict to a process exit code.
+
+    Contract:
+      - Success or no result info → 0
+      - Failure outside a kanban worker → 1
+      - Failure inside a kanban worker (HERMES_KANBAN_TASK env set) WITH
+        failure_reason ∈ {rate_limit, billing} → KANBAN_RATE_LIMIT_EXIT_CODE (75)
+      - Any other failure inside a kanban worker → 1
+
+    Closes hermes-agent#5: the non-quiet `chat -q "..."` path (the one
+    Marvel team workers actually use) previously didn't apply this mapping
+    at all — a rate-limited worker exited 0 by virtue of `cli.chat()`
+    returning cleanly. The dispatcher then classified the 0-exit as a
+    "protocol violation" and auto-blocked the task. This helper is now
+    called from BOTH the quiet (-Q / --quiet) and non-quiet (--q "...")
+    single-query paths so the exit-code contract is consistent.
+    """
+    if not isinstance(result, dict) or not result.get("failed"):
+        return 0
+    if os.environ.get("HERMES_KANBAN_TASK") and result.get(
+        "failure_reason"
+    ) in ("rate_limit", "billing"):
+        try:
+            from hermes_cli.kanban_db import (
+                KANBAN_RATE_LIMIT_EXIT_CODE as _RL_CODE,
+            )
+            return _RL_CODE
+        except Exception:
+            return 1
+    return 1
+
+
 def main(
     query: str = None,
     q: str = None,
@@ -16118,31 +16180,9 @@ def _signal_handler_q(signum, frame):
                     print(f"\nsession_id: {cli.session_id}", file=sys.stderr)
 
                     # Ensure proper exit code for automation wrappers.
-                    #
-                    # Kanban workers get a special case: when the run failed
-                    # purely because the provider rate-limited / exhausted
-                    # quota (not because the task itself is broken), exit with
-                    # the EX_TEMPFAIL sentinel instead of the generic 1. The
-                    # dispatcher's reap classifier maps that code to a
-                    # ``rate_limited`` exit and releases the task back to
-                    # ``ready`` WITHOUT incrementing the failure counter, so a
-                    # 5-hour quota window can't trip the circuit breaker and
-                    # permanently block the card. Non-kanban runs keep the
-                    # plain 0/1 contract automation wrappers expect.
-                    _exit_code = 0
-                    if isinstance(result, dict) and result.get("failed"):
-                        _exit_code = 1
-                        if os.environ.get("HERMES_KANBAN_TASK") and result.get(
-                            "failure_reason"
-                        ) in ("rate_limit", "billing"):
-                            try:
-                                from hermes_cli.kanban_db import (
-                                    KANBAN_RATE_LIMIT_EXIT_CODE as _RL_CODE,
-                                )
-                                _exit_code = _RL_CODE
-                            except Exception:
-                                _exit_code = 1
-                    sys.exit(_exit_code)
+                    # See `_worker_exit_code_from_result` for the kanban
+                    # EX_TEMPFAIL special case (rate-limit → exit 75).
+                    sys.exit(_worker_exit_code_from_result(result))
 
             # Exit with error code if credentials or agent init fails
             sys.exit(1)
@@ -16168,6 +16208,16 @@ def _signal_handler_q(signum, frame):
             cli._show_security_advisories()
             cli.chat(query, images=single_query_images or None)
             cli._print_exit_summary()
+            # Apply the same kanban-worker exit-code mapping that the quiet
+            # path uses, so a rate-limited run gets EX_TEMPFAIL (75) instead
+            # of the implicit 0 that defaults from a clean function return.
+            # Without this, Codex 429 in a worker exits 0 → dispatcher
+            # classifies as "protocol violation" → auto-blocks the task.
+            # (hermes-agent#5 root cause.)
+            _last_result = getattr(cli, "_last_run_result", None)
+            _exit_code = _worker_exit_code_from_result(_last_result)
+            if _exit_code != 0:
+                sys.exit(_exit_code)
         return
 
     # Run interactive mode

gateway/run.py

@@ -5744,6 +5744,63 @@ def _add(path: str) -> None:
                     path, exc,
                 )
 
+    def _write_dispatcher_heartbeat(
+        self,
+        *,
+        heartbeat_path: Path,
+        interval: float,
+        cycles_since_start: int,
+        cycle_started_at: float,
+        any_spawned: bool,
+        spawned_total: int,
+        ready_pending: bool,
+        bad_ticks: int,
+        cycle_error: str | None,
+    ) -> None:
+        """Write a one-shot heartbeat snapshot to disk for external observers.
+
+        Called at the end of every dispatcher cycle (success OR failure).
+        External tools (`hermes gateway status`, monitoring scripts) read
+        this to detect silent stalls — the gateway PID may stay alive but
+        the dispatch loop can stop cycling (hermes-agent#6, #7).
+
+        Write is atomic (temp + rename) and never raises — the caller wraps
+        in try/except so heartbeat-write failures cannot kill the dispatcher.
+
+        Schema (stable; tools may grow tolerant of new fields):
+            schema_version: int (currently 1)
+            last_cycle_ts: float (unix seconds, end of cycle)
+            last_cycle_iso: str (UTC ISO 8601)
+            cycle_started_at: float (unix seconds, start of cycle)
+            cycle_duration_seconds: float
+            interval_seconds: float (configured cadence)
+            cycles_since_start: int (monotonic counter)
+            any_spawned_this_cycle: bool
+            spawned_total_this_cycle: int (count across all boards)
+            ready_pending: bool (whether the ready queue had work this cycle)
+            consecutive_bad_ticks: int
+            gateway_pid: int
+            cycle_error: str | null (exception text if the cycle errored)
+        """
+        from datetime import timezone as _tz
+        now = time.time()
+        payload = {
+            "schema_version": 1,
+            "last_cycle_ts": now,
+            "last_cycle_iso": datetime.fromtimestamp(now, tz=_tz.utc).isoformat(timespec="seconds").replace("+00:00", "Z"),
+            "cycle_started_at": cycle_started_at,
+            "cycle_duration_seconds": max(0.0, now - cycle_started_at),
+            "interval_seconds": float(interval),
+            "cycles_since_start": cycles_since_start,
+            "any_spawned_this_cycle": any_spawned,
+            "spawned_total_this_cycle": spawned_total,
+            "ready_pending": ready_pending,
+            "consecutive_bad_ticks": bad_ticks,
+            "gateway_pid": os.getpid(),
+            "cycle_error": cycle_error,
+        }
+        atomic_json_write(heartbeat_path, payload)
+
     async def _kanban_dispatcher_watcher(self) -> None:
         """Embedded kanban dispatcher — one tick every `dispatch_interval_seconds`.
 
@@ -5918,6 +5975,13 @@ async def _kanban_dispatcher_watcher(self) -> None:
         HEALTH_WINDOW = 6
         bad_ticks = 0
         last_warn_at = 0
+        # Dispatcher heartbeat — written to disk every cycle so external
+        # observers can detect silent stalls (the gateway PID stays alive
+        # but the dispatch loop has stopped cycling). See `hermes gateway
+        # status`. Path is HERMES_HOME/state/dispatcher_health.json.
+        cycles_since_start = 0
+        last_cycle_started_at = 0.0
+        _heartbeat_path = _hermes_home / "state" / "dispatcher_health.json"
         # Avoid hot-looping corrupt-looking board DBs, but do not suppress
         # same-fingerprint retries forever: transient WAL/open races can
         # surface as "database disk image is malformed" for one tick.
@@ -6187,6 +6251,12 @@ def _auto_decompose_tick() -> int:
             "kanban dispatcher: embedded in gateway (interval=%.1fs)", interval
         )
         while self._running:
+            cycles_since_start += 1
+            last_cycle_started_at = time.time()
+            cycle_error: str | None = None
+            any_spawned = False
+            ready_pending = False
+            spawned_total = 0
             try:
                 # Reap zombie children before per-board work so a board DB
                 # failure cannot block cleanup of unrelated workers.
@@ -6204,10 +6274,10 @@ def _auto_decompose_tick() -> int:
                 if auto_decompose_enabled:
                     await asyncio.to_thread(_auto_decompose_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
+                        spawned_total += len(res.spawned)
                         # Quiet by default — only log when something actually
                         # happened, so an idle gateway stays silent.
                         logger.info(
@@ -6240,9 +6310,42 @@ def _auto_decompose_tick() -> int:
                         last_warn_at = now
             except asyncio.CancelledError:
                 logger.debug("kanban dispatcher: cancelled")
+                # Write a final heartbeat noting the cancellation, then re-raise.
+                self._write_dispatcher_heartbeat(
+                    heartbeat_path=_heartbeat_path,
+                    interval=interval,
+                    cycles_since_start=cycles_since_start,
+                    cycle_started_at=last_cycle_started_at,
+                    any_spawned=any_spawned,
+                    spawned_total=spawned_total,
+                    ready_pending=ready_pending,
+                    bad_ticks=bad_ticks,
+                    cycle_error="cancelled",
+                )
                 raise
-            except Exception:
+            except Exception as e:
                 logger.exception("kanban dispatcher: unexpected watcher error")
+                cycle_error = f"{type(e).__name__}: {e}"
+
+            # Write heartbeat every cycle — success OR exception.
+            # External observers (hermes gateway status / monitoring) read this
+            # to detect silent stalls where the gateway PID is alive but the
+            # dispatcher has stopped cycling.
+            try:
+                self._write_dispatcher_heartbeat(
+                    heartbeat_path=_heartbeat_path,
+                    interval=interval,
+                    cycles_since_start=cycles_since_start,
+                    cycle_started_at=last_cycle_started_at,
+                    any_spawned=any_spawned,
+                    spawned_total=spawned_total,
+                    ready_pending=ready_pending,
+                    bad_ticks=bad_ticks,
+                    cycle_error=cycle_error,
+                )
+            except Exception:
+                # Heartbeat-write failure must NEVER kill the dispatcher.
+                logger.exception("kanban dispatcher: heartbeat write failed (non-fatal)")
 
             # Sleep in 1s slices so shutdown is snappy — otherwise a stop()
             # waits up to `interval` seconds for the current sleep to finish.