feat(code): add cache and context status row (#5408)

Depends on #5407

The status footer now shows cache reads and writes, context usage, and
cumulative cost on a second row.

---

Cache metrics reuse pricing normalization and remain scoped to the
active thread, including in-flight and offload usage.

Made by [Open
SWE](https://openswe.vercel.app/agents/019fee35-60e2-723d-ab02-fd45c20e28ed)

## References
- Plan:
https://openswe.vercel.app/agents/019fee35-60e2-723d-ab02-fd45c20e28ed/plan

## Screenshots
<img width="749" height="135" alt="Screenshot 2026-08-11 at 2 49 02 PM"
src="https://github.com/user-attachments/assets/849eedc5-c260-4ca1-807f-6dc78513aaff"
/>

---------

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>

Author: johannes117

libs/code/deepagents_code/_session_stats.py

@@ -173,6 +173,12 @@ class SessionStats:
     output_tokens: int = 0
     """Cumulative output tokens across all LLM requests."""
 
+    cache_read_tokens: int = 0
+    """Cumulative prompt tokens served from provider caches."""
+
+    cache_write_tokens: int = 0
+    """Cumulative prompt tokens written to provider caches."""
+
     total_cost_usd: float = 0.0
     """Cumulative estimated USD cost across priceable LLM requests."""
 
@@ -202,6 +208,8 @@ def record_request(
         *,
         cost_usd: float | None = None,
         kind: UsageKind = "assistant",
+        cache_read_tokens: int = 0,
+        cache_write_tokens: int = 0,
     ) -> None:
         """Accumulate usage for one completed LLM request.
 
@@ -223,10 +231,14 @@ def record_request(
 
                 Missing estimates leave monetary totals unchanged.
             kind: Request class used for `/cost` type breakdowns.
+            cache_read_tokens: Input tokens served from provider caches.
+            cache_write_tokens: Input tokens written to provider caches.
         """
         self.request_count += 1
         self.input_tokens += input_toks
         self.output_tokens += output_toks
+        self.cache_read_tokens += cache_read_tokens
+        self.cache_write_tokens += cache_write_tokens
         if cost_usd is not None:
             self.total_cost_usd += cost_usd
             self.priced_request_count += 1
@@ -276,6 +288,8 @@ def retract_request(self, recorded: RecordedRequest) -> None:
         self.request_count -= 1
         self.input_tokens -= input_toks
         self.output_tokens -= output_toks
+        self.cache_read_tokens -= recorded.cache_read_tokens
+        self.cache_write_tokens -= recorded.cache_write_tokens
         if cost_usd is not None:
             self.total_cost_usd -= cost_usd
             self.priced_request_count -= 1
@@ -318,6 +332,8 @@ def merge(self, other: SessionStats) -> None:
         self.request_count += other.request_count
         self.input_tokens += other.input_tokens
         self.output_tokens += other.output_tokens
+        self.cache_read_tokens += other.cache_read_tokens
+        self.cache_write_tokens += other.cache_write_tokens
         self.total_cost_usd += other.total_cost_usd
         self.priced_request_count += other.priced_request_count
         self.wall_time_seconds += other.wall_time_seconds
@@ -364,6 +380,12 @@ class RecordedRequest:
     output_tokens: int
     """Running output tokens recorded so far for the request."""
 
+    cache_read_tokens: int
+    """Running cache-read tokens recorded so far for the request."""
+
+    cache_write_tokens: int
+    """Running cache-write tokens recorded so far for the request."""
+
     cost_usd: float | None
     """Estimate for the whole request so far, or `None` when unpriceable."""
 
@@ -654,13 +676,17 @@ def _move_request_to_named_model(
         provider,
         cost_usd=cost_usd,
         kind=previous.kind,
+        cache_read_tokens=previous.cache_read_tokens,
+        cache_write_tokens=previous.cache_write_tokens,
     )
     recorded_requests[request_id] = RecordedRequest(
         model_name=model_name,
         provider=provider,
         kind=previous.kind,
         input_tokens=previous.input_tokens,
         output_tokens=previous.output_tokens,
+        cache_read_tokens=previous.cache_read_tokens,
+        cache_write_tokens=previous.cache_write_tokens,
         cost_usd=cost_usd,
         usage_metadata=previous.usage_metadata,
         finalized=previous.finalized,
@@ -782,7 +808,7 @@ def record_message_usage(
                 )
         return None
 
-    from deepagents_code.cost_tracking import estimate_cost
+    from deepagents_code.cost_tracking import cache_token_counts, estimate_cost
 
     model_name, provider = _resolve_usage_model(
         message,
@@ -816,6 +842,8 @@ def record_message_usage(
     # when the fallback is unpriceable, leave a priceable request showing no
     # cost at all.
     cost_usd = estimate_cost(accumulated_usage, model_name, provider)
+    cache_reads, cache_writes = cache_token_counts(accumulated_usage)
+    cache_write_tokens = sum(cache_writes)
 
     stats.record_request(
         model_name,
@@ -824,6 +852,8 @@ def record_message_usage(
         provider,
         cost_usd=cost_usd,
         kind=kind,
+        cache_read_tokens=cache_reads,
+        cache_write_tokens=cache_write_tokens,
     )
     if request_id is not None:
         recorded_requests[request_id] = RecordedRequest(
@@ -832,6 +862,8 @@ def record_message_usage(
             kind=kind,
             input_tokens=input_count,
             output_tokens=output_count,
+            cache_read_tokens=cache_reads,
+            cache_write_tokens=cache_write_tokens,
             cost_usd=cost_usd,
             usage_metadata=accumulated_usage,
             finalized=not is_chunk,

libs/code/deepagents_code/app.py

@@ -4651,6 +4651,7 @@ async def _post_paint_init(self) -> None:
         self._ui_adapter._on_tokens_show = self._show_tokens
         self._ui_adapter._on_session_cost = self._set_session_cost
         self._ui_adapter._on_provisional_cost = self._add_provisional_cost
+        self._ui_adapter._on_usage_update = self._refresh_cache_display
         self._ui_adapter._on_stream_complete = self._mark_thread_turn_completed
 
         if self._server_startup_deferred:
@@ -7642,6 +7643,21 @@ def _show_pending_tokens(self) -> None:
         if self._status_bar:
             self._status_bar.show_pending_tokens()
 
+    def _refresh_cache_display(self) -> None:
+        """Show active-thread cache totals and hit rate, including in-flight use."""
+        if self._status_bar is None:
+            return
+        inputs = self._thread_stats.input_tokens
+        reads = self._thread_stats.cache_read_tokens
+        writes = self._thread_stats.cache_write_tokens
+        if self._inflight_turn_stats is not None and (
+            self._inflight_thread_id == self._lc_thread_id
+        ):
+            inputs += self._inflight_turn_stats.input_tokens
+            reads += self._inflight_turn_stats.cache_read_tokens
+            writes += self._inflight_turn_stats.cache_write_tokens
+        self._status_bar.set_cache_tokens(reads, writes, input_tokens=inputs)
+
     def _set_session_cost(
         self,
         cost_usd: float,
@@ -7720,6 +7736,7 @@ def _reset_thread_usage(
             has_restored_model_usage: Whether restored history contains model usage.
         """
         self._thread_stats = SessionStats()
+        self._refresh_cache_display()
         self._thread_restored_cost_usd = _coerce_session_cost_usd(cost_usd)
         self._thread_has_restored_model_usage = (
             has_restored_model_usage or self._thread_restored_cost_usd > 0
@@ -15388,6 +15405,7 @@ async def _drain(stream_input: Any) -> list[tuple[str, dict[str, Any]]]:  # noqa
             self._session_stats.merge(offload_stats)
             if offload_thread_id == self._lc_thread_id:
                 self._thread_stats.merge(offload_stats)
+                self._refresh_cache_display()
 
     async def _remove_offload_artifacts(
         self,
@@ -15788,6 +15806,7 @@ def _sync_status_model(self) -> None:
             logger.debug("Screen stack empty during model sync", exc_info=True)
         if self._status_bar is None:
             return
+        self._status_bar.set_context_limit(settings.model_context_limit)
         if not provider or not model:
             logger.warning(
                 "Settings missing model identity at status sync "
@@ -16340,6 +16359,7 @@ def _record_goal_grading_run(event: RubricEvaluationEnd) -> None:
                     self._thread_stats.merge(turn_stats)
                 self._inflight_turn_stats = None
                 self._inflight_thread_id = None
+                self._refresh_cache_display()
             # Settle the display on the committed total. Only a completed turn
             # is read back: `durability="exit"` may drop an aborted turn's
             # writes, and the streamed totals already seen are closer to what

libs/code/deepagents_code/app.tcss

@@ -146,8 +146,9 @@ Screen {
 
 /* Status bar */
 #status-bar {
-    height: 1;
+    height: 2;
     dock: bottom;
+    padding: 0;
 }
 
 /* Tool approval widgets */

libs/code/deepagents_code/cost_tracking.py

@@ -256,6 +256,23 @@ def _clamp_cache_counts(
     )
 
 
+def cache_token_counts(
+    usage_metadata: Mapping[str, Any] | None,
+) -> tuple[int, tuple[int, int, int]]:
+    """Return normalized cache reads and generic, 5m, and 1h writes."""
+    if not usage_metadata:
+        return 0, (0, 0, 0)
+    input_tokens = _token_count(usage_metadata.get("input_tokens"))
+    details = usage_metadata.get("input_token_details")
+    if not isinstance(details, Mapping):
+        return 0, (0, 0, 0)
+    return _clamp_cache_counts(
+        input_tokens,
+        _token_count(details.get("cache_read")),
+        _cache_write_counts(details),
+    )
+
+
 def _clamped_detail(
     value: object,
     total: int,
@@ -1348,9 +1365,7 @@ def estimate_cost(
     # numbers, so say so -- a silent clamp can materially undercount.
     original_cache_read = cache_read_tokens
     original_cache_writes = cache_writes
-    cache_read_tokens, cache_writes = _clamp_cache_counts(
-        input_tokens, cache_read_tokens, cache_writes
-    )
+    cache_read_tokens, cache_writes = cache_token_counts(usage_metadata)
     if (
         cache_read_tokens != original_cache_read
         or cache_writes != original_cache_writes

libs/code/deepagents_code/tui/textual_adapter.py

@@ -776,6 +776,9 @@ def __init__(
         a second authority: every server total replaces what this accumulated.
         """
 
+        self._on_usage_update: Callable[[], None] | None = None
+        """Called after streamed request usage changes."""
+
         self._on_stream_complete: Callable[[], None] | None = None
         """Called only after the agent stream reaches a clean end."""
 
@@ -1834,6 +1837,8 @@ async def _after_automatic_compact() -> None:
                             ),
                             recorded_requests=recorded_usage_requests,
                         )
+                    if recorded_usage is not None and adapter._on_usage_update:
+                        adapter._on_usage_update()
                     if recorded_usage is not None and (
                         recorded_usage.cost_usd is not None
                         and adapter._on_provisional_cost