feat(gateway): extend opt-in runtime footer metadata

agent/account_usage.py

@@ -2,6 +2,8 @@
 
 import logging
 import math
+from decimal import Decimal, InvalidOperation
+from urllib.parse import urlparse
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING, Any, Callable, Optional
@@ -385,7 +387,8 @@ def _get_json(url: str, headers: dict[str, str], *, timeout: float) -> dict:
 
 
 def _usage_windows(
-    source: dict, mapping: tuple[tuple[str, str], ...], used_key: str, reset_key: str, *, fraction: bool = False
+    source: dict, mapping: tuple[tuple[str, str], ...], used_key: str, reset_key: str, *, fraction: bool = False,
+    label_fn: Optional[Callable[[dict, str], str]] = None,
 ) -> list[AccountUsageWindow]:
     """Build windows from ``source[key][used_key]``; ``fraction`` scales values <= 1 to percent."""
     windows: list[AccountUsageWindow] = []
@@ -397,7 +400,7 @@ def _usage_windows(
         used = float(used)
         if fraction and used <= 1:
             used *= 100
-        windows.append(AccountUsageWindow(label=label, used_percent=used, reset_at=_parse_dt(window.get(reset_key))))
+        windows.append(AccountUsageWindow(label=label_fn(window, label) if label_fn else label, used_percent=used, reset_at=_parse_dt(window.get(reset_key))))
     return windows
 
 
@@ -587,7 +590,7 @@ def redeem_codex_reset_credit(
 def _fetch_anthropic_account_usage(
     base_url: Optional[str] = None, api_key: Optional[str] = None
 ) -> Optional[AccountUsageSnapshot]:
-    token = (resolve_anthropic_token() or "").strip()
+    token = (str(api_key or "").strip() or (resolve_anthropic_token() or "").strip())
     if not token:
         return None
     if not _is_oauth_token(token):
@@ -645,9 +648,104 @@ def _data(path: str) -> dict:
     return _snapshot("openrouter", "credits_api", windows, details)
 
 
+def _money_symbol(currency: str) -> str:
+    normalized = currency.strip().upper()
+    if normalized == "CNY":
+        return "¥"
+    if normalized == "USD":
+        return "$"
+    return f"{normalized} " if normalized else ""
+
+
+def _decimal_or_none(value: Any) -> Optional[Decimal]:
+    try:
+        parsed = Decimal(str(value).strip())
+    except (InvalidOperation, ValueError, TypeError):
+        return None
+    return parsed if parsed.is_finite() else None
+
+
+def _format_money(value: Decimal, currency: str) -> str:
+    return f"{_money_symbol(currency)}{value:.2f}"
+
+
+def _is_deepseek_base_url(base_url: Optional[str]) -> bool:
+    try:
+        host = urlparse(str(base_url or "")).hostname or ""
+    except Exception:
+        return False
+    host = host.lower().strip(".")
+    return host == "deepseek.com" or host.endswith(".deepseek.com")
+
+
+def _resolve_deepseek_balance_url(base_url: Optional[str]) -> str:
+    parsed = urlparse(str(base_url or "").strip() or "https://api.deepseek.com")
+    scheme = parsed.scheme or "https"
+    netloc = parsed.netloc or parsed.path or "api.deepseek.com"
+    return f"{scheme}://{netloc.rstrip('/')}/user/balance"
+
+
+def _fetch_deepseek_account_usage(base_url: Optional[str], api_key: Optional[str]) -> Optional[AccountUsageSnapshot]:
+    if api_key:
+        runtime = {
+            "base_url": (base_url or "https://api.deepseek.com").strip(),
+            "api_key": str(api_key).strip(),
+        }
+    else:
+        runtime = resolve_runtime_provider(
+            requested="deepseek",
+            explicit_base_url=base_url,
+            explicit_api_key=api_key,
+        )
+    token = str(runtime.get("api_key", "") or "").strip()
+    if not token:
+        return None
+    resolved_base_url = str(runtime.get("base_url", "") or base_url or "https://api.deepseek.com")
+    headers = {
+        "Authorization": f"Bearer {token}",
+        "Accept": "application/json",
+    }
+    with httpx.Client(timeout=10.0) as client:
+        response = client.get(_resolve_deepseek_balance_url(resolved_base_url), headers=headers)
+        response.raise_for_status()
+    payload = response.json() or {}
+    details: list[str] = []
+    for info in payload.get("balance_infos") or []:
+        if not isinstance(info, dict):
+            continue
+        currency = str(info.get("currency") or "").strip().upper()
+        total = _decimal_or_none(info.get("total_balance"))
+        if total is None:
+            continue
+        parts = [f"Balance: {_format_money(total, currency)}"]
+        if currency:
+            parts[0] += f" {currency}"
+        subparts: list[str] = []
+        granted = _decimal_or_none(info.get("granted_balance"))
+        topped_up = _decimal_or_none(info.get("topped_up_balance"))
+        if granted is not None and granted > 0:
+            subparts.append(f"granted {_format_money(granted, currency)}")
+        if topped_up is not None and topped_up > 0:
+            subparts.append(f"topped up {_format_money(topped_up, currency)}")
+        if subparts:
+            parts.append(f"({', '.join(subparts)})")
+        details.append(" ".join(parts))
+    unavailable_reason = None
+    if payload.get("is_available") is False and not details:
+        unavailable_reason = "DeepSeek API balance is insufficient."
+    return AccountUsageSnapshot(
+        provider="deepseek",
+        source="balance_api",
+        fetched_at=_utc_now(),
+        title="Account balance",
+        details=tuple(details),
+        unavailable_reason=unavailable_reason,
+    )
+
+
 _USAGE_FETCHERS: dict[str, Callable[[Optional[str], Optional[str]], Optional[AccountUsageSnapshot]]] = {
     "openai-codex": _fetch_codex_account_usage, "anthropic": _fetch_anthropic_account_usage,
-    "openrouter": _fetch_openrouter_account_usage,
+    "openrouter": _fetch_openrouter_account_usage, "deepseek": _fetch_deepseek_account_usage,
 }
 
 
@@ -675,7 +773,12 @@ def _call_plugin_usage_hook(profile, base_url: Optional[str], api_key: Optional[
 def fetch_account_usage(
     provider: Optional[str], *, base_url: Optional[str] = None, api_key: Optional[str] = None,
 ) -> Optional[AccountUsageSnapshot]:
-    fetcher = _USAGE_FETCHERS.get(str(provider or "").strip().lower())
+    normalized = str(provider or "").strip().lower()
+    if normalized in {"", "auto", "custom"} and not _is_deepseek_base_url(base_url):
+        return None
+    fetcher = _USAGE_FETCHERS.get(normalized)
+    if fetcher is None and _is_deepseek_base_url(base_url):
+        fetcher = _fetch_deepseek_account_usage
     try:
         if fetcher:
             return fetcher(base_url, api_key)

gateway/run_turn.py

@@ -1642,18 +1642,44 @@ def _hmwa_prepend_reasoning(self, agent_result, response, source, _intentional_s
 
     def _hmwa_runtime_footer_line(self, agent_result, source, _turn_seconds):
         """Runtime-metadata footer for the FINAL message of the turn; off by default
-        (display.runtime_footer.enabled=false)."""
+        (display.runtime_footer.enabled=false). Extends the default footer with
+        opt-in provider/account/quota/reasoning fields when configured."""
         from gateway.run import _load_gateway_config, _platform_config_key, _terminal_scope_cwd
         try:
-            from gateway.runtime_footer import build_footer_line as _bfl
+            from gateway.runtime_footer import build_footer_line as _bfl, resolve_footer_config as _rfc
+            _user_config = _load_gateway_config()
+            _platform_key = _platform_config_key(source.platform)
+            _account_usage = agent_result.get("account_usage")
+            _account_label = None
+            if _account_usage is not None:
+                _account_label = (
+                    getattr(_account_usage, "account_label", None)
+                    or getattr(_account_usage, "plan", None)
+                )
+            # Usage is resolved by the producer in run_turn_runner.py while the
+            # routed profile scope and live credential are still available.
+            _footer_cfg = agent_result.get("footer_config") or _rfc(_user_config, _platform_key)
+            _reasoning_effort = agent_result.get("reasoning_effort")
+            if _reasoning_effort is None:
+                _reasoning_cfg = getattr(self, "_reasoning_config", None)
+                if isinstance(_reasoning_cfg, dict):
+                    if _reasoning_cfg.get("enabled") is False:
+                        _reasoning_effort = "none"
+                    else:
+                        _reasoning_effort = _reasoning_cfg.get("effort")
             return _bfl(
-                user_config=_load_gateway_config(),
-                platform_key=_platform_config_key(source.platform), model=agent_result.get("model"),
+                user_config=_user_config,
+                platform_key=_platform_key, model=agent_result.get("model"),
                 context_tokens=agent_result.get("last_prompt_tokens", 0) or 0,
                 context_length=agent_result.get("context_length") or None,
                 cwd=_terminal_scope_cwd(""), turn_seconds=_turn_seconds,
                 requested_model=agent_result.get("requested_model"),
                 served_model=agent_result.get("served_model"),
+                provider=agent_result.get("provider"),
+                account_label=_account_label,
+                account_usage=_account_usage,
+                reasoning_effort=_reasoning_effort,
+                resolved_config=_footer_cfg,
             )
         except Exception as _footer_err:
             logger.debug("runtime_footer build failed: %s", _footer_err)

gateway/run_turn_runner.py

@@ -72,6 +72,38 @@ class _ExecApprovalDeclined(RuntimeError):
     """
 
 
+def _resolve_runtime_footer_metadata(agent, user_config: dict | None, platform_key: str) -> dict:
+    """Resolve footer usage while the turn's routed profile scope is active.
+
+    The API credential is used only to key/schedule the background usage fetch;
+    it is deliberately not returned in the result consumed by ``run_turn.py``.
+    """
+    from gateway.runtime_footer import resolve_footer_config
+    from gateway.runtime_footer_usage import get_cached
+    from hermes_constants import get_hermes_home
+
+    footer_config = resolve_footer_config(user_config, platform_key)
+    fields = set(footer_config.get("fields") or ())
+    needs_usage = footer_config.get("enabled") and bool(fields & {"account", "quota"})
+    provider = getattr(agent, "provider", None) if agent is not None else None
+    base_url = getattr(agent, "base_url", None) if agent is not None else None
+    api_key = getattr(agent, "api_key", None) if agent is not None else None
+    account_usage = None
+    if needs_usage and provider:
+        account_usage = get_cached(
+            provider,
+            base_url=base_url,
+            api_key=api_key,
+            hermes_home=str(get_hermes_home()),
+        )
+    return {
+        "provider": provider,
+        "base_url": base_url,
+        "account_usage": account_usage,
+        "footer_config": footer_config,
+    }
+
+
 class TurnRunner:
     """Per-turn collaborator carrying ``GatewayRunner._run_agent_inner``'s tool-progress callbacks."""
 
@@ -1968,6 +2000,14 @@ def run_sync(self):
             "model": getattr(agent, "model", None) if agent else None,
             "context_length": (getattr(comp, "context_length", 0) or 0) if has_comp else 0,
         }
+        footer_metadata = _resolve_runtime_footer_metadata(
+            agent,
+            ctx.user_config,
+            platform_key,
+        )
+        footer_metadata["reasoning_effort"] = (
+            getattr(runner, "_reasoning_config", {}) or {}
+        ).get("effort")
         compacted_in_place, effective_session_id, history_offset = self._sync_session_after_run(agent_history)
         # failure_reason must survive the empty-response path too (TUI billing, transient-failure
         # persistence). compression_deferred (soft lock-contention defer) is distinct from
@@ -1984,6 +2024,7 @@ def run_sync(self):
             "tools": ctx.tools_holder[0] or [],
             "history_offset": history_offset, "compacted_in_place": compacted_in_place, "session_id": effective_session_id,
             **usage,
+            **footer_metadata,
         }
         if not final_response:
             final_response = _normalize_empty_agent_response(result, final_response or "", history_len=len(agent_history))