feat(gateway): emit token_usage events after each API call for real-time context bar

agent/conversation_loop.py

@@ -1841,6 +1841,19 @@ def _stop_spinner():
                     agent.session_cache_write_tokens += canonical_usage.cache_write_tokens
                     agent.session_reasoning_tokens += canonical_usage.reasoning_tokens
 
+                    # Emit structured token usage for real-time context bar updates.
+                    # Fire-and-forget: the target may not have a status_callback wired,
+                    # and we never want to block the conversation loop.
+                    agent._emit_token_usage(
+                        input_tokens=agent.session_input_tokens,
+                        output_tokens=agent.session_output_tokens,
+                        total_tokens=agent.session_total_tokens,
+                        context_tokens=agent.context_compressor.last_prompt_tokens
+                        if agent.context_compressor else 0,
+                        context_length=agent.context_compressor.context_length
+                        if agent.context_compressor else 0,
+                    )
+
                     # Log API call details for debugging/observability
                     _cache_pct = ""
                     if canonical_usage.cache_read_tokens and prompt_tokens:

run_agent.py

@@ -768,6 +768,31 @@ def _emit_warning(self, message: str) -> None:
             except Exception:
                 logger.debug("status_callback error in _emit_warning", exc_info=True)
 
+    def _emit_token_usage(self, input_tokens: int, output_tokens: int,
+                          total_tokens: int, context_tokens: int,
+                          context_length: int) -> None:
+        """Emit structured token usage data to the gateway for real-time UI updates.
+
+        The gateway forwards this as a structured ``token.usage`` event so the
+        desktop app's context bar can update smoothly during a turn rather than
+        jumping to the final value at end-of-turn.
+        """
+        if not self.status_callback:
+            return
+        import json
+        payload = json.dumps({
+            "input_tokens": input_tokens,
+            "output_tokens": output_tokens,
+            "total_tokens": total_tokens,
+            "context_tokens": context_tokens,
+            "context_length": context_length,
+            "context_pct": round(context_tokens / context_length * 100, 1) if context_length > 0 else 0,
+        })
+        try:
+            self.status_callback("token_usage", payload)
+        except Exception:
+            logger.debug("status_callback error in _emit_token_usage", exc_info=True)
+
     # ── Buffered retry/fallback status ────────────────────────────────────
     # Retry and fallback chains were flooding the CLI/gateway with status
     # noise that users found confusing: a single transient 429 could produce

tests/test_tui_gateway_server.py

@@ -4929,6 +4929,28 @@ def test_make_agent_handles_null_agent_config(monkeypatch):
     assert mock_agent.call_args.kwargs["max_iterations"] == 80
 
 
+def test_make_agent_passes_fallback_chain_from_config(monkeypatch):
+    fallback_chain = [
+        {"provider": "opencode-zen", "model": "deepseek-v4-flash-free"},
+        {"provider": "taro", "model": "qwen3.6-27b-256k"},
+    ]
+    _setup_make_agent_mocks(
+        monkeypatch,
+        {
+            "fallback_providers": fallback_chain,
+            "fallback_model": {
+                "provider": "opencode-zen",
+                "model": "deepseek-v4-flash-free",
+            },
+        },
+    )
+
+    with patch("run_agent.AIAgent") as mock_agent:
+        server._make_agent("sid1", "key1")
+
+    assert mock_agent.call_args.kwargs["fallback_model"] == fallback_chain
+
+
 class _FakeAgentForBackground:
     base_url = None
     api_key = None
@@ -4983,6 +5005,20 @@ def test_background_agent_kwargs_handles_null_agent_config(monkeypatch):
     assert kwargs["max_iterations"] == 40
 
 
+def test_background_agent_kwargs_preserves_full_fallback_chain(monkeypatch):
+    monkeypatch.setattr(server, "_load_cfg", lambda: {})
+    agent = _FakeAgentForBackground()
+    agent._fallback_chain = [
+        {"provider": "opencode-zen", "model": "deepseek-v4-flash-free"},
+        {"provider": "taro", "model": "qwen3.6-27b-256k"},
+    ]
+    agent._fallback_model = {"provider": "opencode-zen", "model": "legacy-only"}
+
+    kwargs = server._background_agent_kwargs(agent, "task_1")
+
+    assert kwargs["fallback_model"] == agent._fallback_chain
+
+
 def test_config_show_displays_nested_max_turns(monkeypatch):
     monkeypatch.setattr(
         server,

tui_gateway/server.py

@@ -17,6 +17,7 @@
 
 from hermes_constants import get_hermes_home
 from hermes_cli.env_loader import load_hermes_dotenv
+from hermes_cli.fallback_config import get_fallback_chain
 from utils import is_truthy_value
 from tui_gateway.transport import (
     StdioTransport,
@@ -394,6 +395,16 @@ def _status_update(sid: str, kind: str, text: str | None = None):
     body = (text if text is not None else kind).strip()
     if not body:
         return
+    # Structured token_usage events are JSON-encoded; unpack and re-emit
+    # as a typed event so the desktop app can consume them directly.
+    if kind == "token_usage":
+        import json
+        try:
+            payload = json.loads(body)
+        except (json.JSONDecodeError, TypeError):
+            return
+        _emit("token.usage", sid, payload)
+        return
     _emit(
         "status.update",
         sid,
@@ -1972,7 +1983,10 @@ def _background_agent_kwargs(agent, task_id: str) -> dict:
         "request_overrides": dict(getattr(agent, "request_overrides", {}) or {}),
         "platform": "tui",
         "session_db": _get_db(),
-        "fallback_model": getattr(agent, "_fallback_model", None),
+        "fallback_model": (
+            list(getattr(agent, "_fallback_chain", None) or [])
+            or getattr(agent, "_fallback_model", None)
+        ),
     }
 
 
@@ -2019,6 +2033,7 @@ def _make_agent(sid: str, key: str, session_id: str | None = None):
         pass
 
     cfg = _load_cfg()
+    fallback_chain = get_fallback_chain(cfg)
     agent_cfg = cfg.get("agent") or {}
     system_prompt = (agent_cfg.get("system_prompt", "") or "").strip()
     startup_skills = _parse_tui_skills_env()
@@ -2050,6 +2065,7 @@ def _make_agent(sid: str, key: str, session_id: str | None = None):
         acp_command=runtime.get("command"),
         acp_args=runtime.get("args"),
         credential_pool=runtime.get("credential_pool"),
+        fallback_model=fallback_chain or None,
         quiet_mode=True,
         # verbose_logging controls DEBUG-level agent logging; it is intentionally
         # independent of tool_progress_mode (which only controls per-tool