feat(gateway): add per-turn token usage footer to gateway responses

gateway/run.py

@@ -68,6 +68,56 @@
 # ``config.yaml`` ``agent.gateway_auto_continue_freshness``.
 _AUTO_CONTINUE_FRESHNESS_SECS_DEFAULT = 60 * 60
 
+# Strip the model's own `usage:t=...` line before appending the gateway's
+# runtime footer.  The agent system prompt instructs models to emit a usage
+# line; when the gateway also injects token information in the runtime footer
+# (snapshot-diff values), we get two different numbers in the same reply.
+# Matching against the agent-prompt format (backtick-delimited, comma-separated
+# thousands, optional cached/reasoning suffixes) and stripping it avoids the
+# duplicate.  (issue #17592 / PR discussion)
+_AGENT_USAGE_LINE_RE = re.compile(
+    r'\n*`usage:t=[\d,]+ i=[\d,]+(?: \(\+ [\d,]+ c\))? o=[\d,]+(?: \(r [\d,]+\))?`[ \t]*$'
+)
+
+
+def _snapshot_agent_token_usage(agent: Any) -> "Dict[str, int]":
+    """Return cumulative token counters from an AIAgent instance."""
+    if agent is None:
+        return {}
+    keys = (
+        "input_tokens", "output_tokens", "total_tokens",
+        "cache_read_tokens", "cache_write_tokens", "reasoning_tokens",
+        "prompt_tokens", "completion_tokens",
+    )
+    attr_map = {
+        "input_tokens": "session_input_tokens",
+        "output_tokens": "session_output_tokens",
+        "total_tokens": "session_total_tokens",
+        "cache_read_tokens": "session_cache_read_tokens",
+        "cache_write_tokens": "session_cache_write_tokens",
+        "reasoning_tokens": "session_reasoning_tokens",
+        "prompt_tokens": "session_prompt_tokens",
+        "completion_tokens": "session_completion_tokens",
+    }
+    snapshot: Dict[str, int] = {}
+    for key in keys:
+        try:
+            snapshot[key] = int(getattr(agent, attr_map[key], 0) or 0)
+        except (TypeError, ValueError):
+            snapshot[key] = 0
+    return snapshot
+
+
+def _diff_agent_token_usage(
+    before: "Dict[str, int]", after: "Dict[str, int]"
+) -> "Dict[str, int]":
+    """Return non-negative per-turn token usage deltas."""
+    keys = set(before or {}) | set(after or {})
+    return {
+        key: max(0, int((after or {}).get(key, 0)) - int((before or {}).get(key, 0)))
+        for key in keys
+    }
+
 
 def _coerce_gateway_timestamp(value: Any) -> Optional[float]:
     """Best-effort conversion of stored gateway timestamps to epoch seconds.
@@ -5296,11 +5346,19 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
                     context_tokens=agent_result.get("last_prompt_tokens", 0) or 0,
                     context_length=agent_result.get("context_length") or None,
                     cwd=os.environ.get("TERMINAL_CWD", ""),
+                    token_usage=agent_result.get("token_usage") or {
+                        "input_tokens": agent_result.get("input_tokens", 0) or 0,
+                        "output_tokens": agent_result.get("output_tokens", 0) or 0,
+                    },
                 )
             except Exception as _footer_err:
                 logger.debug("runtime_footer build failed: %s", _footer_err)
                 _footer_line = ""
             if _footer_line and response and not agent_result.get("already_sent"):
+                # Strip the agent's self-reported usage line before the gateway
+                # appends its own token footer (they differ: the agent-side line
+                # is the model's estimate; ours is per-turn snapshot-diff).
+                response = _AGENT_USAGE_LINE_RE.sub('', response).rstrip()
                 response = f"{response}\n\n{_footer_line}"
 
             # Emit agent:end hook
@@ -10970,6 +11028,7 @@ def _approval_notify_sync(approval_data: dict) -> None:
             _approval_session_key = session_key or ""
             _approval_session_token = set_current_session_key(_approval_session_key)
             register_gateway_notify(_approval_session_key, _approval_notify_sync)
+            _usage_before = _snapshot_agent_token_usage(agent)
             try:
                 # If _prepare_inbound_message_text buffered image paths for native
                 # attachment, wrap the user turn as an OpenAI-style multimodal
@@ -11008,6 +11067,10 @@ def _approval_notify_sync(approval_data: dict) -> None:
                 unregister_gateway_notify(_approval_session_key)
                 reset_current_session_key(_approval_session_token)
             result_holder[0] = result
+            _turn_usage = _diff_agent_token_usage(
+                _usage_before,
+                _snapshot_agent_token_usage(agent),
+            )
 
             # Signal the stream consumer that the agent is done
             if _stream_consumer is not None:
@@ -11044,6 +11107,7 @@ def _approval_notify_sync(approval_data: dict) -> None:
                     "output_tokens": _output_toks,
                     "model": _resolved_model,
                     "context_length": _context_length,
+                    "token_usage": _turn_usage,
                 }
 
             # Scan tool results for MEDIA:<path> tags that need to be delivered
@@ -11149,6 +11213,7 @@ def _approval_notify_sync(approval_data: dict) -> None:
                 "output_tokens": _output_toks,
                 "model": _resolved_model,
                 "context_length": _context_length,
+                "token_usage": _turn_usage,
                 "session_id": effective_session_id,
                 "response_previewed": result.get("response_previewed", False),
             }

gateway/runtime_footer.py

@@ -27,12 +27,62 @@
 
 import os
 from pathlib import Path
-from typing import Any, Iterable, Optional
+from typing import Any, Dict, Iterable, Optional
 
 _DEFAULT_FIELDS: tuple[str, ...] = ("model", "context_pct", "cwd")
 _SEP = " · "
 
 
+def _format_usage_int(value: Any) -> str:
+    """Format an integer with thousands separators; return '0' on failure."""
+    try:
+        return f"{max(0, int(value)):,}"
+    except (TypeError, ValueError):
+        return "0"
+
+
+def _render_token_usage(token_usage: Optional[Dict[str, int]]) -> str:
+    """Build the compact token-usage string: ``usage:t=<N> i=<N> o=<N>``.
+
+    Cached read tokens are appended as ``(+ N c)``, reasoning tokens as
+    ``(r N)``.  Returns ``\"\"`` when *token_usage* is empty or missing.
+    """
+    usage = token_usage or {}
+    input_tokens = usage.get("input_tokens", usage.get("prompt_tokens", 0)) or 0
+    output_tokens = usage.get("output_tokens", usage.get("completion_tokens", 0)) or 0
+
+    try:
+        display_total = int(input_tokens) + int(output_tokens)
+    except (TypeError, ValueError):
+        display_total = 0
+
+    if display_total <= 0:
+        return ""
+
+    parts = [
+        f"usage:t={_format_usage_int(display_total)}",
+        f"i={_format_usage_int(input_tokens)}",
+    ]
+
+    try:
+        cached = int(usage.get("cache_read_tokens", 0) or 0)
+    except (TypeError, ValueError):
+        cached = 0
+    if cached > 0:
+        parts[-1] = f"{parts[-1]} (+ {_format_usage_int(cached)} c)"
+
+    output_part = f"o={_format_usage_int(output_tokens)}"
+    try:
+        reasoning = int(usage.get("reasoning_tokens", 0) or 0)
+    except (TypeError, ValueError):
+        reasoning = 0
+    if reasoning > 0:
+        output_part = f"{output_part} (r {_format_usage_int(reasoning)})"
+    parts.append(output_part)
+
+    return " ".join(parts)
+
+
 def _home_relative_cwd(cwd: str) -> str:
     """Return *cwd* with ``$HOME`` collapsed to ``~``.  Empty string if unset."""
     if not cwd:
@@ -95,6 +145,7 @@ def format_runtime_footer(
     context_tokens: int,
     context_length: Optional[int],
     cwd: Optional[str] = None,
+    token_usage: Optional[Dict[str, int]] = None,
     fields: Iterable[str] = _DEFAULT_FIELDS,
 ) -> str:
     """Render the footer line, or return "" if no fields have data.
@@ -116,11 +167,15 @@ def format_runtime_footer(
             rel = _home_relative_cwd(cwd or os.environ.get("TERMINAL_CWD", ""))
             if rel:
                 parts.append(rel)
+        elif field == "tokens":
+            rendered = _render_token_usage(token_usage)
+            if rendered:
+                parts.append(rendered)
         # Unknown field names are silently ignored.
 
     if not parts:
         return ""
-    return _SEP.join(parts)
+    return f"`{_SEP.join(parts)}`"
 
 
 def build_footer_line(
@@ -131,6 +186,7 @@ def build_footer_line(
     context_tokens: int,
     context_length: Optional[int],
     cwd: Optional[str] = None,
+    token_usage: Optional[Dict[str, int]] = None,
 ) -> str:
     """Top-level entry point used by gateway/run.py.
 
@@ -146,5 +202,6 @@ def build_footer_line(
         context_tokens=context_tokens,
         context_length=context_length,
         cwd=cwd,
+        token_usage=token_usage,
         fields=cfg.get("fields") or _DEFAULT_FIELDS,
     )