feat(code): add /context usage report

libs/code/COMMANDS.md

@@ -8,7 +8,7 @@ Regenerate this file with `make commands-catalog` after changing command names,
 aliases, descriptions, visibility, or hidden-command metadata.
 
 
-## Public (38)
+## Public (39)
 
 | Command | Aliases | Description |
 | --- | --- | --- |
@@ -18,6 +18,7 @@ aliases, descriptions, visibility, or hidden-command metadata.
 | `/auto-update` |  | Turn automatic updates on or off |
 | `/changelog` |  | Open the changelog in a browser |
 | `/clear` |  | Clear the chat and start a new thread |
+| `/context` |  | Show current context window usage |
 | `/copy` |  | Copy the latest assistant message to clipboard |
 | `/cost` |  | Show estimated thread cost |
 | `/docs` |  | Open the docs |

libs/code/deepagents_code/app.py

@@ -109,6 +109,7 @@
 )
 from deepagents_code.tui.widgets._links import open_url_async
 from deepagents_code.tui.widgets.chat_input import ChatInput
+from deepagents_code.tui.widgets.context_usage import ContextUsageScreen
 from deepagents_code.tui.widgets.goal_status import GoalStatusPanel
 from deepagents_code.tui.widgets.loading import LoadingWidget
 from deepagents_code.tui.widgets.message_store import (
@@ -14109,6 +14110,27 @@ async def _handle_command(self, command: str) -> None:
                 timeout=5,
                 markup=False,
             )
+        elif cmd == "/context":
+            context_tokens, conversation_tokens = await self._get_context_usage_counts()
+            if context_tokens is None and not self._tokens_approximate:
+                context_tokens = self._context_tokens or None
+            approximate = context_tokens is None and bool(
+                conversation_tokens or self._context_tokens
+            )
+            if approximate:
+                conversation_tokens = conversation_tokens or self._context_tokens
+            elif context_tokens is None:
+                context_tokens = 0
+            self.push_screen(
+                ContextUsageScreen(
+                    context_tokens=context_tokens,
+                    conversation_tokens=conversation_tokens,
+                    context_limit=settings.model_context_limit,
+                    model_spec=self._effective_model_spec() or settings.model_name,
+                    approximate=approximate,
+                ),
+                lambda _result: self._focus_chat_input_after_refresh(),
+            )
         elif cmd == "/tokens":
             await self._mount_message(UserMessage(command))
             if self._context_tokens > 0:
@@ -14804,32 +14826,40 @@ async def _has_conversation_messages(self) -> bool:
             )
             return True
 
-    async def _get_conversation_token_count(self) -> int | None:
-        """Return the approximate conversation-only token count.
+    async def _get_context_usage_counts(self) -> tuple[int | None, int | None]:
+        """Read provider-reported total and estimated conversation usage together.
 
         Returns:
-            Token count as an integer, or `None` if state is unavailable.
+            Pair of provider-reported context tokens and approximate effective
+                conversation tokens. Either value is `None` when unavailable.
         """
-        if not self._agent:
-            return None
+        if not self._agent or not self._lc_thread_id:
+            return None, None
         try:
-            from langchain_core.messages.utils import (
-                count_tokens_approximately,
+            from langchain_core.messages.utils import count_tokens_approximately
+
+            values = await self._get_thread_state_values(self._lc_thread_id)
+            reported = _persisted_context_tokens(values) or None
+            messages = values.get("messages", [])
+            if not isinstance(messages, list) or not messages:
+                return reported, None
+            effective = _effective_conversation(
+                messages,
+                values.get("_summarization_event"),
             )
+            return reported, count_tokens_approximately(effective)
+        except Exception:  # best-effort for context-usage displays
+            logger.debug("Failed to retrieve context usage", exc_info=True)
+            return None, None
 
-            config: RunnableConfig = {
-                "configurable": {"thread_id": self._lc_thread_id},
-            }
-            state = await self._agent.aget_state(config)
-            if not state or not state.values:
-                return None
-            messages = state.values.get("messages", [])
-            if not messages:
-                return None
-            return count_tokens_approximately(messages)
-        except Exception:  # best-effort for /tokens display
-            logger.debug("Failed to retrieve conversation token count", exc_info=True)
-            return None
+    async def _get_conversation_token_count(self) -> int | None:
+        """Return the approximate conversation-only token count.
+
+        Returns:
+            Token count as an integer, or `None` if state is unavailable.
+        """
+        _, conversation = await self._get_context_usage_counts()
+        return conversation
 
     async def _handle_offload(self) -> None:
         """Offload older messages to free context window space.
@@ -15051,7 +15081,7 @@ async def _handle_offload(self) -> None:
                     )
                 )
 
-            self._on_tokens_update(tokens_after)
+            self._on_tokens_update(tokens_after, approximate=True)
 
         except Exception as exc:  # surface offload errors to user
             logger.exception("Offload failed")

libs/code/deepagents_code/command_registry.py

@@ -127,6 +127,12 @@ def to_entry(self) -> CommandEntry:
         description="Copy the latest assistant message to clipboard",
         bypass_tier=BypassTier.SIDE_EFFECT_FREE,
     ),
+    SlashCommand(
+        name="/context",
+        description="Show current context window usage",
+        bypass_tier=BypassTier.QUEUED,
+        hidden_keywords="tokens window usage remaining offload compact",
+    ),
     SlashCommand(
         name="/cost",
         description="Show estimated thread cost",

libs/code/deepagents_code/tui/widgets/context_usage.py

@@ -0,0 +1,169 @@
+"""Color-coded context-window visualization for `/context`."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, ClassVar
+
+from textual.binding import Binding, BindingType
+from textual.containers import VerticalScroll
+from textual.content import Content
+from textual.screen import ModalScreen
+from textual.widgets import Static
+
+from deepagents_code import theme
+from deepagents_code._session_stats import format_token_count
+from deepagents_code.config import get_glyphs, is_ascii_mode
+
+if TYPE_CHECKING:
+    from textual.app import ComposeResult
+
+
+class _ContextUsage(Static):
+    def __init__(
+        self,
+        *,
+        context_tokens: int | None,
+        conversation_tokens: int | None,
+        context_limit: int | None,
+        model_spec: str | None,
+        approximate: bool,
+    ) -> None:
+        super().__init__()
+        self._total = None if context_tokens is None else max(0, context_tokens)
+        self._conversation = (
+            None if conversation_tokens is None else max(0, conversation_tokens)
+        )
+        self._limit = context_limit if context_limit and context_limit > 0 else None
+        self._model = model_spec or "Unknown model"
+        self._approximate = approximate
+
+    def render(self) -> Content:
+        colors = theme.get_theme_colors(self)
+        glyphs = get_glyphs()
+        usage = self._total if self._total is not None else self._conversation or 0
+        maximum = format_token_count(self._limit) if self._limit else "unavailable"
+        prefix = "~" if self._approximate or self._total is None else ""
+        right = f"{prefix}{format_token_count(usage)} / {maximum}"
+        if self._total is not None and self._limit:
+            right += f"  {self._total / self._limit * 100:.1f}%"
+        left = Content.assemble(
+            ("Context", f"bold {colors.primary}"),
+            (f" {glyphs.bullet} ", colors.muted),
+            self._model,
+        )
+        gap = self.content_size.width - left.cell_length - len(right)
+        header = Content.assemble(left, " " * max(1, gap), (right, colors.muted))
+
+        categories: list[tuple[str, int, str]] = []
+        if self._total is None:
+            if self._conversation:
+                categories.append(
+                    ("Conversation estimate", self._conversation, colors.primary)
+                )
+        elif self._total:
+            if self._conversation is None:
+                categories.append(("Used context", self._total, colors.secondary))
+            else:
+                conversation = min(self._conversation, self._total)
+                if overhead := self._total - conversation:
+                    categories.append(
+                        ("System prompt + tools", overhead, colors.warning)
+                    )
+                if conversation:
+                    categories.append(("Conversation", conversation, colors.primary))
+        if self._total is not None and self._limit:
+            categories.append(
+                ("Free space", max(0, self._limit - self._total), colors.muted)
+            )
+
+        scale = max(self._limit or 0, sum(tokens for _, tokens, _ in categories), 1)
+        width = max(self.content_size.width, 1)
+        used = 0
+        segments: list[Content] = []
+        for _label, tokens, color in categories:
+            end = round(min(scale, used + tokens) / scale * width)
+            start = round(min(scale, used) / scale * width)
+            segments.append(
+                Content.styled(glyphs.box_horizontal * (end - start), color)
+            )
+            used += tokens
+        bar = Content.assemble(*segments)
+
+        rows: list[Content] = []
+        marker = glyphs.box_horizontal * 2
+        for label, tokens, color in categories:
+            percent = tokens / scale * 100
+            value = f"{format_token_count(tokens)}  {glyphs.bullet}  {percent:.1f}%"
+            item = Content.assemble((marker, color), " ", label)
+            rows.append(
+                Content.assemble(
+                    item,
+                    " " * max(1, width - item.cell_length - len(value)),
+                    (value, colors.muted),
+                )
+            )
+        if not rows:
+            rows.append(Content.styled("No context usage reported yet.", colors.muted))
+        elif self._total is None:
+            rows.append(Content.styled("Total usage unavailable.", colors.muted))
+        return Content("\n").join((header, bar, *rows))
+
+
+class ContextUsageScreen(ModalScreen[None]):
+    """Modal visualization of the current model context window."""
+
+    BINDINGS: ClassVar[list[BindingType]] = [
+        Binding("escape", "close", "Close", show=False)
+    ]
+    CSS = """
+    ContextUsageScreen { align: center middle; }
+    ContextUsageScreen > VerticalScroll {
+        width: 94%; max-width: 120; height: auto; max-height: 90%;
+        background: $surface; border: solid $primary; padding: 1 2;
+    }
+    ContextUsageScreen _ContextUsage { height: auto; }
+    ContextUsageScreen .context-usage-help {
+        height: 1; color: $text-muted; margin-top: 2;
+    }
+    """
+
+    def __init__(
+        self,
+        *,
+        context_tokens: int | None,
+        conversation_tokens: int | None,
+        context_limit: int | None,
+        model_spec: str | None,
+        approximate: bool,
+    ) -> None:
+        """Initialize the modal from the latest usage measurements."""
+        super().__init__()
+        self._usage = _ContextUsage(
+            context_tokens=context_tokens,
+            conversation_tokens=conversation_tokens,
+            context_limit=context_limit,
+            model_spec=model_spec,
+            approximate=approximate,
+        )
+
+    def compose(self) -> ComposeResult:
+        """Compose the context visualization and close hint.
+
+        Yields:
+            Widgets that make up the modal.
+        """
+        with VerticalScroll():
+            yield self._usage
+            yield Static("Esc to close", classes="context-usage-help")
+
+    def on_mount(self) -> None:
+        """Use an ASCII border when the terminal cannot render Unicode."""
+        if is_ascii_mode():
+            self.query_one(VerticalScroll).styles.border = (
+                "ascii",
+                theme.get_theme_colors(self).primary,
+            )
+
+    def action_close(self) -> None:
+        """Dismiss the context visualization."""
+        self.dismiss(None)

libs/code/deepagents_code/tui/widgets/startup_tip.py

@@ -20,6 +20,7 @@
     "Use @ to reference files and / for commands": 3,
     "Try /threads to resume a previous conversation": 2,
     "Use /offload to summarize older messages and free up the context window": 2,
+    "Use /context to see context window usage and remaining space": 1,
     "Use /copy to copy the latest message": 3,
     "Use /cost to see a breakdown of estimated spend": 1,
     "Use /tools to list the tools available to the agent": 1,

libs/code/tests/unit_tests/test_app.py

@@ -179,6 +179,34 @@ def test_strips_provider_prefix(
         assert _display_model_label(spec) == expected
 
 
+async def test_context_prefers_checkpoint_total_after_offload(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    app = DeepAgentsApp()
+    app._context_tokens = 200
+    app._tokens_approximate = True
+    monkeypatch.setattr(
+        app, "_get_context_usage_counts", AsyncMock(return_value=(1_000, 200))
+    )
+    push_screen = MagicMock()
+    monkeypatch.setattr(app, "push_screen", push_screen)
+    focus = MagicMock()
+    monkeypatch.setattr(app, "_focus_chat_input_after_refresh", focus)
+
+    with (
+        patch("deepagents_code.app.ContextUsageScreen") as screen_type,
+        patch("deepagents_code.config.settings") as settings,
+    ):
+        settings.model_provider = "anthropic"
+        settings.model_name = "claude-sonnet"
+        settings.model_context_limit = 2_000
+        await app._handle_command("/context")
+
+    assert screen_type.call_args.kwargs["context_tokens"] == 1_000
+    push_screen.call_args.args[1](None)
+    focus.assert_called_once_with()
+
+
 class TestWhatsNewMessage:
     """Tests for the post-upgrade banner content."""