fix(kanban): preserve quota exits in non-quiet workers

cli.py

@@ -1484,6 +1484,28 @@ def _finalize_single_query(cli) -> None:
         cli._release_active_session()
 
 
+def _kanban_worker_rate_limit_exit_code(result: object) -> Optional[int]:
+    """Return the dispatcher's temporary-failure sentinel on provider quota.
+
+    Kanban workers use both the fully-quiet and human-readable single-query
+    paths. Keep their provider-quota exit contract identical so the dispatcher
+    requeues the task without counting a task failure.
+    """
+    if (
+        not os.environ.get("HERMES_KANBAN_TASK")
+        or not isinstance(result, dict)
+        or not result.get("failed")
+        or result.get("failure_reason") not in ("rate_limit", "billing")
+    ):
+        return None
+
+    try:
+        from hermes_cli.kanban_db import KANBAN_RATE_LIMIT_EXIT_CODE
+    except Exception:
+        return 1
+    return KANBAN_RATE_LIMIT_EXIT_CODE
+
+
 def _reset_terminal_input_modes_on_exit() -> None:
     """Best-effort: disable focus reporting + mouse tracking on TUI exit so they
     don't leak into the next shell session sharing the tab.
@@ -16694,6 +16716,7 @@ def chat(self, message, images: list = None, voice_input: bool = False) -> Optio
         # this to True. Early returns (credential refresh failure, etc.)
         # leave it False, which is correct — those aren't user interrupts.
         self._last_turn_interrupted = False
+        self._kanban_worker_exit_code = None
 
         # Refresh provider credentials if needed (handles key rotation transparently)
         if not self._ensure_runtime_credentials():
@@ -17189,6 +17212,11 @@ def run_agent():
                 # but guard against edge cases.
                 agent_thread.join(timeout=30)
 
+            # Preserve the structured provider outcome before any post-turn
+            # rendering or output flush can fail.  The outer one-shot boundary
+            # consumes this after the normal response and exit summary print.
+            self._kanban_worker_exit_code = _kanban_worker_rate_limit_exit_code(result)
+
             # Freeze per-prompt elapsed timer once the agent thread has
             # exited (or been abandoned as a daemon after interrupt).
             if self._prompt_start_time is not None:
@@ -22072,19 +22100,13 @@ def _signal_handler_q(signum, frame):
                         # 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
+                        _exit_code = _kanban_worker_rate_limit_exit_code(result)
+                        if _exit_code is None:
+                            _exit_code = (
+                                1
+                                if isinstance(result, dict) and result.get("failed")
+                                else 0
+                            )
                         sys.exit(_exit_code)
 
                 # Exit with error code if credentials or agent init fails
@@ -22111,6 +22133,9 @@ def _signal_handler_q(signum, frame):
                 cli._show_security_advisories()
                 cli.chat(query, images=single_query_images or None)
                 cli._print_exit_summary(clear_screen=False)
+                _kanban_exit_code = getattr(cli, "_kanban_worker_exit_code", None)
+                if _kanban_exit_code is not None:
+                    sys.exit(_kanban_exit_code)
         finally:
             _finalize_single_query(cli)
         return

tests/cli/test_cli_interrupt_ack_race.py

@@ -154,6 +154,36 @@ def test_unacknowledged_interrupt_message_is_requeued_not_dropped():
     assert agent.clear_calls >= 1
 
 
+def test_kanban_quota_exit_survives_post_result_render_failure():
+    """A later display failure must not turn provider quota into exit 0."""
+    from hermes_cli.kanban_db import KANBAN_RATE_LIMIT_EXIT_CODE
+
+    cli = _make_cli()
+    agent = _StubAgent(cli.session_id, turn_seconds=0)
+    agent.run_conversation = MagicMock(return_value={
+        "final_response": "",
+        "messages": [],
+        "api_calls": 1,
+        "completed": False,
+        "failed": True,
+        "failure_reason": "rate_limit",
+        "error": "quota exhausted",
+    })
+    cli.agent = agent
+
+    with patch.dict("os.environ", {"HERMES_KANBAN_TASK": "t_rate_limited"}, clear=False), \
+         patch.object(cli, "_ensure_runtime_credentials", return_value=True), \
+         patch.object(cli, "_resolve_turn_agent_config", return_value={
+             "signature": cli._active_agent_route_signature,
+             "model": None, "runtime": None, "request_overrides": None,
+         }), \
+         patch.object(cli, "_init_agent", return_value=True), \
+         patch.object(cli, "_flush_stream", side_effect=RuntimeError("render failed")):
+        assert cli.chat("work kanban task t_rate_limited") is None
+
+    assert cli._kanban_worker_exit_code == KANBAN_RATE_LIMIT_EXIT_CODE
+
+
 
 
 def test_chat_persists_clean_input_when_a_queued_note_changes_api_message():

tests/cli/test_kanban_rate_limit_exit.py

@@ -0,0 +1,95 @@
+"""Kanban workers preserve provider-quota failures as temporary exits."""
+
+import pytest
+
+import cli
+from hermes_cli.kanban_db import KANBAN_RATE_LIMIT_EXIT_CODE
+
+
+@pytest.mark.parametrize("failure_reason", ["rate_limit", "billing"])
+def test_kanban_worker_provider_quota_exits_tempfail(monkeypatch, failure_reason):
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_rate_limited")
+
+    assert (
+        cli._kanban_worker_rate_limit_exit_code(
+            {"failed": True, "failure_reason": failure_reason}
+        )
+        == KANBAN_RATE_LIMIT_EXIT_CODE
+    )
+
+
+@pytest.mark.parametrize(
+    "result",
+    [
+        {"failed": False, "failure_reason": "rate_limit"},
+        {"failed": True, "failure_reason": "provider_error"},
+        None,
+    ],
+)
+def test_kanban_worker_non_quota_result_keeps_normal_exit(monkeypatch, result):
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_not_rate_limited")
+
+    assert cli._kanban_worker_rate_limit_exit_code(result) is None
+
+
+def test_non_kanban_quota_failure_keeps_normal_exit(monkeypatch):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+
+    assert (
+        cli._kanban_worker_rate_limit_exit_code(
+            {"failed": True, "failure_reason": "rate_limit"}
+        )
+        is None
+    )
+
+
+def test_human_readable_worker_prints_summary_before_tempfail(monkeypatch):
+    calls = []
+
+    class FakeCLI:
+        def __init__(self, **_kwargs):
+            self.session_id = "kanban-worker-session"
+            self.agent = type(
+                "Agent",
+                (),
+                {"session_id": self.session_id, "platform": "cli"},
+            )()
+            self.console = type(
+                "Console",
+                (),
+                {"print": lambda _self, *args, **kwargs: calls.append(
+                    ("console", args, kwargs)
+                )},
+            )()
+            self._kanban_worker_exit_code = None
+
+        def _claim_active_session(self, surface, *, stderr=False):
+            calls.append(("claim", surface, stderr))
+            return True
+
+        def _show_security_advisories(self):
+            calls.append("advisories")
+
+        def chat(self, query, images=None):
+            calls.append(("chat", query, images))
+            self._kanban_worker_exit_code = KANBAN_RATE_LIMIT_EXIT_CODE
+
+        def _print_exit_summary(self, clear_screen=True):
+            calls.append(("summary", clear_screen))
+
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_rate_limited")
+    monkeypatch.setattr(cli, "HermesCLI", FakeCLI)
+    monkeypatch.setattr(cli.atexit, "register", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(
+        cli,
+        "_finalize_single_query",
+        lambda fake_cli: calls.append(("finalize", fake_cli.session_id)),
+    )
+
+    with pytest.raises(SystemExit) as exc_info:
+        cli.main(query="work kanban task t_rate_limited", quiet=False)
+
+    assert exc_info.value.code == KANBAN_RATE_LIMIT_EXIT_CODE
+    assert calls.index(("summary", False)) < calls.index(
+        ("finalize", "kanban-worker-session")
+    )