fix(cli): propagate a failed one-shot turn's exit code on the non-quiet route

cli.py

@@ -2537,6 +2537,9 @@ class HermesCLI(CLIProcessNotificationsMixin, CLIAgentSetupMixin, CLICommandsMix
     # Seeded -q first message (see _should_seed_interactive); run() re-creates
     # _pending_input, so it is enqueued only after the fresh queue exists.
     _seeded_first_message: Optional["_SeededQueryMessage"] = None
+    # Last turn's result dict (set by chat()); one-shot callers read it for the
+    # automation exit code. Optionally-typed on partially built instances.
+    _last_turn_result: Optional[dict] = None
     # Inspection surfaces (banner, /tools, status line) read this on partially built instances too.
     disabled_toolsets: Optional[List[str]] = None
 
@@ -4111,19 +4114,38 @@ def _run_quiet_single_query(cli, effective_query):
 
     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
-    sys.exit(_exit_code)
+    # Exit code 0/1 for automation wrappers, EX_TEMPFAIL for a kanban quota wall.
+    # Shared with the non-quiet one-shot route so BOTH propagate a failed turn —
+    # rc=0 on a failed turn reads as a clean exit to the kanban dispatcher.
+    sys.exit(_single_query_exit_code(result))
+
+
+def _single_query_exit_code(result) -> int:
+    """Automation exit code for one one-shot turn.
+
+    A failed turn MUST NOT exit 0: the kanban dispatcher books ``rc=0`` on a
+    still-``running`` task as a *protocol violation* ("worker exited cleanly without
+    calling kanban_complete or kanban_block"), which hides the real cause and burns the
+    violation retry budget on a provider crash. Shared by the quiet (``-Q``) and
+    non-quiet (``chat -q`` — the shape the dispatcher spawns) one-shot routes.
+
+    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).
+    """
+    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 as _RL_CODE
+
+            return _RL_CODE
+        except Exception:
+            return 1
+    return 1
 
 
 def _route_single_query_images(cli, query, effective_query, single_query_images, single_query_image_urls):
@@ -4441,6 +4463,13 @@ def _run_single_query_mode(cli, query, image, quiet, oneshot):
         cli._show_security_advisories()
         cli.chat(query, images=single_query_images or None)
         cli._print_exit_summary(clear_screen=False)
+        # Same exit-code contract as the -Q route: a failed turn must never exit 0
+        # (the kanban dispatcher reads that as a protocol violation). ``chat`` returns
+        # None when credentials/agent init failed before any turn ran — that is a 1 too.
+        _turn_result = getattr(cli, "_last_turn_result", None)
+        if _turn_result is None:
+            sys.exit(1)
+        sys.exit(_single_query_exit_code(_turn_result))
     finally:
         _finalize_single_query(cli)
 

hermes_cli/cli_chat_turn_mixin.py

@@ -87,6 +87,9 @@ def chat(self, message, images: list = None, voice_input: bool = False) -> Optio
             agent_thread.start()
             interrupt_msg = self._chat_monitor_agent_thread(turn, agent_thread)
             self._chat_settle_turn(turn)
+            # Retained for one-shot callers: the turn result drives the process exit
+            # code (a failed turn must never exit 0). ``chat()`` itself returns text only.
+            self._last_turn_result = turn.result
             return self._chat_render_turn(turn, agent_thread, interrupt_msg)
         except Exception as e:
             print(f"Error: {e}")

tests/hermes_cli/test_chat_q_exit_clear.py

@@ -86,6 +86,8 @@ def _show_security_advisories(self):
 
         def chat(self, query, images=None):
             calls.append(("chat", query, images))
+            # Production chat() retains the turn result: the one-shot route exits with it.
+            self._last_turn_result = {"completed": True, "failed": False}
             return "done"
 
         def _print_exit_summary(self, clear_screen=True):
@@ -101,8 +103,10 @@ def _print_exit_summary(self, clear_screen=True):
         lambda fake_cli: calls.append(("finalize", fake_cli.session_id)),
     )
 
-    cli_mod.main(query="hello", quiet=False, toolsets="terminal")
+    with pytest.raises(SystemExit) as exc_info:
+        cli_mod.main(query="hello", quiet=False, toolsets="terminal")
 
+    assert exc_info.value.code == 0  # success still exits 0 on the one-shot route
     assert calls == [
         ("claim", "cli", False),
         "query-label",

tests/hermes_cli/test_single_query_exit_code.py

@@ -0,0 +1,98 @@
+"""A failed one-shot turn must never exit 0.
+
+The kanban dispatcher books ``rc=0`` on a still-``running`` task as a *protocol
+violation* (worker exited without ``kanban_complete``/``kanban_block``) — which is
+wrong twice over: the work was usually in flight, and a provider/quota wall is not
+a task failure at all. Only the ``-Q`` (quiet) one-shot branch propagated a failure
+exit code, so every non-quiet worker (the dispatcher spawns ``hermes -p <profile>
+--cli chat -q …``) reported a mid-flight provider death as "worker exited cleanly
+(rc=0) without calling kanban_complete or kanban_block".
+
+These pin the behaviour contract of the shared exit-code decision and of the
+non-quiet route that must use it.
+"""
+
+from __future__ import annotations
+
+import cli as cli_mod
+from hermes_cli.kanban_db import KANBAN_RATE_LIMIT_EXIT_CODE
+
+
+class _StubCLI:
+    """Minimum surface ``_run_single_query_mode`` touches on the non-quiet route."""
+
+    def __init__(self, turn_result, response="done"):
+        self._last_turn_result = turn_result
+        self._response = response
+        self.chatted = False
+        self.exit_summary_printed = False
+
+        class _Console:
+            def print(self, *a, **k):
+                pass
+
+        self.console = _Console()
+
+    def _claim_active_session(self, *a, **k):
+        return True
+
+    def _show_security_advisories(self):
+        pass
+
+    def chat(self, query, images=None):
+        self.chatted = True
+        return self._response
+
+    def _print_exit_summary(self, **kwargs):
+        self.exit_summary_printed = True
+
+
+def _run_nonquiet(monkeypatch, turn_result, response="done"):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    monkeypatch.setattr(cli_mod, "_finalize_single_query", lambda _cli: None)
+    stub = _StubCLI(turn_result, response=response)
+    try:
+        cli_mod._run_single_query_mode(stub, "work kanban task t_x", None, False, True)
+    except SystemExit as exc:
+        return stub, exc.code
+    return stub, None
+
+
+def test_failed_turn_is_nonzero_off_kanban(monkeypatch):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    assert cli_mod._single_query_exit_code({"failed": True, "error": "boom"}) == 1
+
+
+def test_clean_turn_is_zero(monkeypatch):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    assert cli_mod._single_query_exit_code({"completed": True, "failed": False}) == 0
+    assert cli_mod._single_query_exit_code(None) == 0
+
+
+def test_kanban_quota_wall_keeps_tempfail_sentinel(monkeypatch):
+    """A rate-limit/billing wall must not be booked as a task failure."""
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_test")
+    for reason in ("rate_limit", "billing"):
+        assert cli_mod._single_query_exit_code(
+            {"failed": True, "failure_reason": reason}
+        ) == KANBAN_RATE_LIMIT_EXIT_CODE
+    assert cli_mod._single_query_exit_code(
+        {"failed": True, "failure_reason": "api_error"}
+    ) == 1
+
+
+def test_nonquiet_one_shot_propagates_failed_turn(monkeypatch):
+    """The route the kanban dispatcher spawns must exit nonzero on a failed turn."""
+    stub, code = _run_nonquiet(monkeypatch, {"failed": True, "error": "provider down"})
+    assert stub.chatted, "the non-quiet route must still run the turn"
+    assert stub.exit_summary_printed, "the exit summary is part of the route's contract"
+    assert code == 1, (
+        f"a failed non-quiet one-shot turn exited {code!r}: rc=0 is read by the kanban "
+        "dispatcher as a protocol violation, hiding provider crashes"
+    )
+
+
+def test_nonquiet_one_shot_still_exits_zero_on_success(monkeypatch):
+    stub, code = _run_nonquiet(monkeypatch, {"completed": True, "failed": False})
+    assert stub.chatted
+    assert code == 0, f"a successful one-shot turn must exit 0, got {code!r}"

tests/hermes_cli/test_single_query_session_finalize.py

@@ -108,6 +108,9 @@ def _show_security_advisories(self):
 
         def chat(self, query, images=None):
             calls.append(("chat", query, images))
+            # Production chat() retains the turn result; the one-shot route reads it
+            # for the process exit code.
+            self._last_turn_result = {"completed": True, "failed": False}
             return "done"
 
         def _print_exit_summary(self, clear_screen=True):
@@ -121,8 +124,12 @@ def _print_exit_summary(self, clear_screen=True):
         lambda fake_cli: calls.append(("finalize", fake_cli.session_id)),
     )
 
-    cli_mod.main(query="hello", quiet=False, toolsets="terminal")
+    with pytest.raises(SystemExit) as exc_info:
+        cli_mod.main(query="hello", quiet=False, toolsets="terminal")
 
+    # A successful non-quiet one-shot turn still exits 0; the session is finalized on
+    # the way out (the exit lives inside the try/finally).
+    assert exc_info.value.code == 0
     assert calls == [
         ("claim", "cli", False),
         "query-label",