fix(cli): one-shot -q runs report their outcome in the exit code

The Kanban dispatcher spawns workers as `hermes ... chat -q <prompt>`
(`kanban_db.py::_default_spawn`). That path ran the turn and fell through
to an implicit 0 whatever happened — success, failure, or a provider
quota wall.

`detect_crashed_workers` reads rc=0 with the task still `running` as a
protocol violation, and protocol violations trip the breaker at
`failure_limit=1`, so a single HTTP 429 blocked the card permanently and
every card queued behind it stayed in `todo` forever waiting on a parent
that could never reach `done`.

`KANBAN_RATE_LIMIT_EXIT_CODE` (EX_TEMPFAIL) exists precisely to prevent
this: `_classify_worker_exit` maps it to a `rate_limited` kind and the
task is released back to `ready` without counting a failure. The consumer
end was complete and tested. The producer end was wired into the `-Q`
path only — the one the dispatcher does not use.

This extracts that mapping into `_single_query_exit_code()` and applies it
on both one-shot paths. `chat()` returns the rendered response string, so
the non-quiet path could not see the outcome; `_chat_settle_turn` now
records the raw turn result for it to read.

Scope is deliberately narrow. The non-quiet path only exits non-zero when
`HERMES_KANBAN_TASK` is set, so interactive runs and ordinary `hermes chat
-q` invocations still exit 0 exactly as before. For a dispatcher-spawned
worker the full contract now applies: 0 on success, 1 on failure, and the
sentinel on a rate-limit/billing wall.

Tests cover the path that was missed rather than the one that already
worked: 16 of the 17 new assertions fail on the parent commit, and the
key regression fails as `assert None == 75` — the exact rc=0 fall-through
— rather than on a missing symbol. The seventeenth asserts that a human's
one-shot run keeps exiting 0, and passes both before and after.

cli.py

@@ -4089,6 +4089,31 @@ def _sync_cli_session_id_from_agent(cli) -> None:
         cli.session_id = cli.agent.session_id
 
 
+def _single_query_exit_code(result) -> int:
+    """Map a one-shot turn result onto a process exit code.
+
+    0 success, 1 failure, and ``KANBAN_RATE_LIMIT_EXIT_CODE`` (EX_TEMPFAIL) when a
+    Kanban worker failed purely because the provider rate-limited or the account hit a
+    billing/quota wall. The dispatcher's reap classifier maps that sentinel to a
+    ``rate_limited`` exit and releases the task back to ``ready`` WITHOUT counting a
+    failure, so a multi-day quota window cannot trip the circuit breaker and
+    permanently block the card.
+
+    Shared by both one-shot paths. It previously lived inline in the ``-Q`` path only,
+    which is how the ``-q`` path — the one the Kanban dispatcher actually spawns —
+    ended up with no exit contract at all.
+    """
+    if not (isinstance(result, dict) and 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
+            return KANBAN_RATE_LIMIT_EXIT_CODE
+        except Exception:
+            return 1
+    return 1
+
+
 def _run_quiet_single_query(cli, effective_query, emitter=None):
     """Quiet (-Q) one-shot turn: run, print the response (stderr for errors/session_id), then sys.exit with the automation exit code.
     With a ``StreamJsonEmitter`` the final answer and the exit line become the terminal ``result`` JSONL record instead.
@@ -4174,18 +4199,7 @@ def _run_quiet_single_query(cli, effective_query, emitter=None):
     if emitter is None:
         print(f"\nsession_id: {cli.session_id}", file=sys.stderr)
 
-    # Exit code 0/1 for automation wrappers. Kanban workers that failed purely on
-    # rate-limit/billing exit with the EX_TEMPFAIL sentinel so the dispatcher releases
-    # the task without counting a failure (a quota window must not trip the breaker).
-    _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
+    _exit_code = _single_query_exit_code(result)
     if emitter is not None:
         _exit_code = emitter.emit_result(result, session_id=cli.session_id or "", exit_code=_exit_code)
     sys.exit(_exit_code)
@@ -4529,6 +4543,14 @@ def _run_single_query_mode(cli, query, image, quiet, oneshot, stream_json: bool
         cli._show_security_advisories()
         cli.chat(query, images=single_query_images or None)
         cli._print_exit_summary(clear_screen=False)
+        # A dispatcher-spawned Kanban worker must report its outcome in its exit code.
+        # This path fell through to an implicit 0 for every outcome, and the reap
+        # classifier reads rc=0 with the task still `running` as a protocol violation,
+        # which blocks the card on the FIRST occurrence. A provider quota wall
+        # therefore killed the card permanently, along with everything queued behind
+        # it. Interactive and plain `-q` runs are unaffected: they still exit 0.
+        if os.environ.get("HERMES_KANBAN_TASK"):
+            sys.exit(_single_query_exit_code(getattr(cli, "_last_turn_result", None)))
     finally:
         _finalize_single_query(cli)
 

hermes_cli/cli_chat_turn_mixin.py

@@ -22,6 +22,11 @@ from typing import Optional
 class CLIChatTurnMixin:
     """chat() and its per-turn phase helpers."""
 
+    # Last completed turn's raw agent result. chat() returns only the rendered
+    # response string, so one-shot callers that must map an outcome onto a
+    # process exit code (see cli._run_single_query_mode) read this instead.
+    _last_turn_result = None
+
     def chat(self, message, images: list = None, voice_input: bool = False) -> Optional[str]:
         """Run one user turn; returns the agent's response, or None on error.
 
@@ -441,6 +446,7 @@ class CLIChatTurnMixin:
             self._prompt_duration = max(0.0, time.time() - self._prompt_start_time)
             self._prompt_start_time = None
         self._last_turn_finished_at = time.time()  # status bar idle time
+        self._last_turn_result = turn.result
         # AsyncOpenAI clients bound to the worker's now-closed loop would crash
         # prompt_toolkit's loop from __del__ on GC.
         try:

tests/hermes_cli/test_single_query_exit_contract.py

@@ -0,0 +1,154 @@
+"""One-shot runs map their outcome onto an exit code, on BOTH one-shot paths.
+
+The Kanban dispatcher spawns workers as ``hermes ... chat -q <prompt>`` — the
+non-quiet single-query path. That path had no exit contract at all: it ran the
+turn and fell through to an implicit 0 whatever happened. The dispatcher's reap
+classifier reads rc=0 with the task still ``running`` as a protocol violation
+and blocks the card on the first occurrence, so one provider quota wall killed
+the card permanently along with every card queued behind it.
+
+The EX_TEMPFAIL sentinel that exists to prevent exactly that was wired into the
+``-Q`` path only. These tests pin the contract on both.
+"""
+
+from __future__ import annotations
+
+import os
+from types import SimpleNamespace
+
+import pytest
+
+import cli
+from hermes_cli.kanban_db import KANBAN_RATE_LIMIT_EXIT_CODE
+
+
+# --------------------------------------------------------------------------
+# the shared mapping
+# --------------------------------------------------------------------------
+
+@pytest.fixture(autouse=True)
+def _no_inherited_kanban_env(monkeypatch):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    monkeypatch.delenv("HERMES_KANBAN_GOAL_MODE", raising=False)
+
+
+@pytest.mark.parametrize("result", [
+    {"final_response": "ok"},
+    {"failed": False},
+    None,
+    "not a dict",
+])
+def test_a_run_that_did_not_fail_exits_zero(result):
+    assert cli._single_query_exit_code(result) == 0
+
+
+def test_a_plain_failure_exits_one():
+    assert cli._single_query_exit_code({"failed": True, "failure_reason": "tool_error"}) == 1
+
+
+@pytest.mark.parametrize("reason", ["rate_limit", "billing"])
+def test_a_kanban_worker_on_a_quota_wall_exits_with_the_sentinel(monkeypatch, reason):
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_abc123")
+    result = {"failed": True, "failure_reason": reason}
+    assert cli._single_query_exit_code(result) == KANBAN_RATE_LIMIT_EXIT_CODE
+
+
+@pytest.mark.parametrize("reason", ["rate_limit", "billing"])
+def test_the_sentinel_is_for_kanban_workers_only(reason):
+    """A human's one-shot run that hit a quota wall is an ordinary failure."""
+    assert "HERMES_KANBAN_TASK" not in os.environ
+    assert cli._single_query_exit_code({"failed": True, "failure_reason": reason}) == 1
+
+
+def test_a_kanban_worker_failing_for_another_reason_still_exits_one(monkeypatch):
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_abc123")
+    assert cli._single_query_exit_code({"failed": True, "failure_reason": "tool_error"}) == 1
+
+
+# --------------------------------------------------------------------------
+# the path the dispatcher actually spawns: `chat -q`, non-quiet
+# --------------------------------------------------------------------------
+
+def _fake_cli(turn_result):
+    """A CLI stub exercising only what the non-quiet one-shot tail touches."""
+    return SimpleNamespace(
+        _single_query_mode=False,
+        _claim_active_session=lambda *a, **k: True,
+        console=SimpleNamespace(print=lambda *a, **k: None),
+        _show_security_advisories=lambda: None,
+        chat=lambda *a, **k: "response",
+        _print_exit_summary=lambda **k: None,
+        _last_turn_result=turn_result,
+    )
+
+
+def _run_non_quiet(monkeypatch, turn_result):
+    monkeypatch.setattr(cli, "_should_seed_interactive", lambda *a, **k: False)
+    monkeypatch.setattr(cli, "_collect_query_images", lambda q, i: (q, []))
+    monkeypatch.setattr(cli, "_collect_kanban_task_images", lambda imgs: [])
+    monkeypatch.setattr(cli, "_finalize_single_query", lambda c: None)
+    stub = _fake_cli(turn_result)
+    try:
+        cli._run_single_query_mode(stub, "work kanban task t_abc123", None, False, True)
+    except SystemExit as exc:
+        return exc.code
+    return None  # fell through without exiting
+
+
+@pytest.mark.parametrize("reason", ["rate_limit", "billing"])
+def test_dispatcher_spawned_worker_signals_a_quota_wall_not_a_protocol_violation(monkeypatch, reason):
+    """The regression. rc=0 here is read as a protocol violation and blocks the card."""
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_abc123")
+    code = _run_non_quiet(monkeypatch, {"failed": True, "failure_reason": reason})
+    assert code == KANBAN_RATE_LIMIT_EXIT_CODE
+
+
+def test_dispatcher_spawned_worker_reports_an_ordinary_failure_as_nonzero(monkeypatch):
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_abc123")
+    code = _run_non_quiet(monkeypatch, {"failed": True, "failure_reason": "tool_error"})
+    assert code == 1
+
+
+def test_dispatcher_spawned_worker_that_succeeded_exits_zero(monkeypatch):
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_abc123")
+    code = _run_non_quiet(monkeypatch, {"final_response": "done"})
+    assert code == 0
+
+
+def test_a_human_one_shot_run_is_unaffected(monkeypatch):
+    """No HERMES_KANBAN_TASK: the path must not start exiting non-zero on people."""
+    code = _run_non_quiet(monkeypatch, {"failed": True, "failure_reason": "rate_limit"})
+    assert code is None
+
+
+# --------------------------------------------------------------------------
+# the plumbing that makes the outcome reachable from that path
+# --------------------------------------------------------------------------
+
+def test_settling_a_turn_records_the_raw_result():
+    """chat() returns a rendered string; the exit mapping needs the result dict."""
+    from hermes_cli.cli_chat_turn_mixin import CLIChatTurnMixin
+
+    stub = SimpleNamespace(
+        _prompt_start_time=None,
+        _prompt_duration=0.0,
+        _flush_stream=lambda: None,
+        conversation_history=[],
+        agent=None,
+    )
+    result = {"failed": True, "failure_reason": "rate_limit"}
+    turn = SimpleNamespace(
+        result=result, use_streaming_tts=False, text_queue=None, tts_thread=None,
+    )
+
+    CLIChatTurnMixin._chat_settle_turn(stub, turn)
+
+    assert stub._last_turn_result is result
+
+
+def test_the_recorded_result_defaults_to_none():
+    """A CLI that never completed a turn must map to 0, not raise."""
+    from hermes_cli.cli_chat_turn_mixin import CLIChatTurnMixin
+
+    assert CLIChatTurnMixin._last_turn_result is None
+    assert cli._single_query_exit_code(CLIChatTurnMixin._last_turn_result) == 0