langchain-ai/deepagents
1
0
Fork 0
mirror of https://github.com/langchain-ai/deepagents.git synced 2026-07-21 17:25:26 -04:00
Code Issues 169 Packages Projects Releases 16 Wiki Activity

feat(code): simplify welcome banner to compact box (#4482)

Browse Source 
The startup welcome banner was a large multi-section panel — ASCII art
title, a "Ready to code!" subheader, a randomly selected tip, LangSmith
tracing links (including replica projects), editable-install path, and
MCP status. On a fresh launch with an empty chat, Textual's
bottom-anchor behavior floated this panel to the bottom of the viewport,
leaving dead space above it — unlike Claude Code, which greets the user
top-aligned.

This PR replaces the banner with a compact bordered box and fixes the
viewport anchoring so the banner sits at the top on launch.

The new box shows `dcode` (bold) and the version tag. Rows below the
title appear only when their data is present:

    dcode  v0.1.0

    tracing:   'shared-deepagents-code'
    mcp:       42 tools

The model and working-directory rows are opt-in (`SPLASH_SHOW_MODEL` /
`SPLASH_SHOW_CWD`, both off by default) — the model already lives in the
status bar and the cwd in the prompt. The thread ID row appears only in
debug mode. MCP server warnings (unauthenticated, errored, awaiting
reconnect) still surface as yellow alert lines inside the box, and the
editable-install path shows for local installs.

Removed: rotating tips and the `build_welcome_footer()` helper, the
"Ready to code!" subheader, replica tracing project display, and
`HIDE_SPLASH_TIPS`. `set_idle()` / `set_connecting()` are now no-ops —
the status bar owns connection progress and failure messaging.

`_ChatScroll.anchor()` no longer bottom-anchors at startup. The real
anchor is deferred until chat content overflows the viewport, so the
banner stays pinned to the top on an empty chat and the chat follows the
bottom once it fills up. Streaming, shell output, slash-command output,
and thread-resume paths opt into bottom-follow as content arrives.

`_sync_status_model` now updates the banner's model display (when
enabled) in addition to the status bar, so `/model` switches are
reflected everywhere. Cwd changes also flow through `update_cwd()`. Both
lookups guard against `ScreenStackError` when the banner isn't mounted.

## Release note

The startup welcome banner is now a compact bordered box instead of a
multi-section panel. Tips, the "Ready to code!" subheader, and replica
tracing project lines have been removed. The banner stays top-aligned in
an empty chat instead of floating to the bottom. Two new opt-in settings
— `SPLASH_SHOW_MODEL` and `SPLASH_SHOW_CWD` (env:
`DEEPAGENTS_CODE_SPLASH_SHOW_MODEL` / `DEEPAGENTS_CODE_SPLASH_SHOW_CWD`)
— control whether the active model and working directory appear as rows
in the banner.

---------

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
Co-authored-by: Mason Daugherty <mason@langchain.dev>
Co-authored-by: Mason Daugherty <github@mdrxy.com>
This commit is contained in:
Johannes du Plessis
2026-07-05 20:51:46 -07:00
committed by GitHub
parent 69bb06c068
commit b7f46e9318
6 changed files with 1308 additions and 1066 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/code/deepagents_code/_env_vars.py
+19 -9
View File
@@ -34,11 +34,6 @@ AUTO_UPDATE = "DEEPAGENTS_CODE_AUTO_UPDATE"
"""Toggle automatic app updates. Enabled by default; set to a falsy value
('0', 'false', 'no', 'off', or empty) to opt out."""
DANGEROUSLY_OVERRIDE_STARTUP_SUBHEADER = (
"DEEPAGENTS_CODE_DANGEROUSLY_OVERRIDE_STARTUP_SUBHEADER"
)
"""Override the startup splash subheader text when set."""
DEBUG = "DEEPAGENTS_CODE_DEBUG"
"""Enable verbose debug logging and preserve the server subprocess log.
@@ -102,7 +97,12 @@ EXTRA_SKILLS_DIRS = "DEEPAGENTS_CODE_EXTRA_SKILLS_DIRS"
"""Colon-separated paths added to the skill containment allowlist."""
HIDE_CWD = "DEEPAGENTS_CODE_HIDE_CWD"
"""Hide local path displays in the TUI footer and startup splash when enabled."""
"""Hide local path displays in the TUI footer and the editable-install path in
the startup splash when enabled.
Does not control the splash working-directory row, which is gated solely by
`SPLASH_SHOW_CWD`.
"""
HIDE_GIT_BRANCH = "DEEPAGENTS_CODE_HIDE_GIT_BRANCH"
"""Hide the current git branch in the TUI footer when enabled."""
@@ -110,9 +110,6 @@ HIDE_GIT_BRANCH = "DEEPAGENTS_CODE_HIDE_GIT_BRANCH"
HIDE_LANGSMITH_TRACING = "DEEPAGENTS_CODE_HIDE_LANGSMITH_TRACING"
"""Hide LangSmith tracing project/thread info in the startup splash when enabled."""
HIDE_SPLASH_TIPS = "DEEPAGENTS_CODE_HIDE_SPLASH_TIPS"
"""Hide rotating tips in the startup splash when enabled."""
HIDE_SPLASH_VERSION = "DEEPAGENTS_CODE_HIDE_SPLASH_VERSION"
"""Hide version and local-install details in the splash screen when enabled."""
@@ -225,6 +222,19 @@ Defaults to enabled; set to a falsy value (`0`, `false`, `no`, `off`, or empty)
to suppress the success toast while still opening URLs normally.
"""
SPLASH_SHOW_CWD = "DEEPAGENTS_CODE_SPLASH_SHOW_CWD"
"""Show the working-directory row in the startup welcome banner when enabled.
Off by default and independent of the status bar's `HIDE_CWD`.
"""
SPLASH_SHOW_MODEL = "DEEPAGENTS_CODE_SPLASH_SHOW_MODEL"
"""Show the active model row in the startup welcome banner when enabled.
Off by default; the model is always visible in the status bar, so the banner
row is opt-in to avoid duplicating it.
"""
SUPPRESS_ENV_OVERRIDE_WARNING = "DEEPAGENTS_CODE_SUPPRESS_ENV_OVERRIDE_WARNING"
"""Silence the startup warning emitted when a `DEEPAGENTS_CODE_`-prefixed
LangSmith variable overrides its canonical counterpart (e.g. both
libs/code/deepagents_code/app.py
+160 -36
View File
@@ -330,6 +330,8 @@ if TYPE_CHECKING:
from langgraph.pregel import Pregel
from textual.app import ComposeResult
from textual.events import MouseUp, Paste
from textual.geometry import Size
from textual.layout import DockArrangeResult
from textual.scrollbar import ScrollUp
from textual.timer import Timer
from textual.widget import Widget
@@ -1800,6 +1802,82 @@ class _ChatScroll(VerticalScroll):
FOCUS_ON_CLICK = False
# The deferred-anchor logic below drives the base class through its private
# anchor state (`_anchored`, `_anchor_released`) and mirrors the compositor's
# arrange-then-check ordering. Validated against Textual 8.2.7; a base-class
# rename or reflow change could break it silently, so `TestChatScrollAnchoring`
# is the safety net for Textual upgrades.
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Initialize the chat scroll container.
Sets `_follow_bottom_when_scrollable`, the bottom-follow intent flag that
`arrange()` only honors once content overflows the viewport (see
`anchor()`).
"""
super().__init__(*args, **kwargs)
self._follow_bottom_when_scrollable = False
def anchor(self, anchor: bool = True) -> None:
"""Anchor only once the transcript is tall enough to scroll.
Textual's default bottom anchor also bottom-aligns content that is
shorter than the viewport, which makes the welcome banner snap down as
soon as the first message arrives. Deferring the real anchor preserves
the top-aligned banner while still following the bottom after overflow.
Args:
anchor: When `True`, arm bottom-follow so the view sticks to the
bottom once content overflows (engaging immediately if it
already does). When `False`, disarm bottom-follow entirely and
delegate to the base class.
"""
self._follow_bottom_when_scrollable = anchor
if not anchor:
super().anchor(False)
return
self._anchor_released = False
if self._is_scrollable():
super().anchor(True)
return
super().anchor(False)
self.scroll_y = 0
self.scroll_target_y = 0
def release_anchor(self) -> None:
"""Release bottom-follow intent when the user scrolls manually."""
self._follow_bottom_when_scrollable = False
super().release_anchor()
def arrange(self, size: Size, optimal: bool = False) -> DockArrangeResult:
"""Arrange children and enable bottom-follow only after overflow.
Args:
size: Size of the chat scroll container.
optimal: Whether fr units should avoid expanding widgets.
Returns:
Widget placement information for the arranged children.
"""
result = super().arrange(size, optimal=optimal)
if not self._follow_bottom_when_scrollable or self._anchor_released:
return result
viewport_height = self.container_size.height - self.scrollbar_size_horizontal
if result.spatial_map.total_region.bottom > viewport_height:
self._anchored = True
else:
self._anchored = False
# Reset the scroll offset without firing `watch_scroll_y`, matching
# how the compositor mutates scroll state mid-arrange (avoids a
# redundant refresh cycle from within the layout pass itself).
self.set_reactive(VerticalScroll.scroll_y, 0.0)
self.set_reactive(VerticalScroll.scroll_target_y, 0.0)
return result
def _is_scrollable(self) -> bool:
"""Return whether current chat content overflows the viewport."""
return self.max_scroll_y > 0
class DeepAgentsApp(App):
"""Main Textual application for deepagents-code."""
@@ -2799,6 +2877,7 @@ class DeepAgentsApp(App):
UI components for the main chat area and status bar.
"""
from deepagents_code._env_vars import SHOW_HEADER, is_env_truthy
from deepagents_code.config import settings
if is_env_truthy(SHOW_HEADER) or self._installation_stale:
yield _StaticHeader(id="app-header")
@@ -2807,6 +2886,9 @@ class DeepAgentsApp(App):
# `_ChatScroll` keeps clicks on messages from stealing input focus.
with _ChatScroll(id="chat"):
yield WelcomeBanner(
model_provider=settings.model_provider or "",
model_name=settings.model_name or "",
cwd=self._cwd,
thread_id=self._lc_thread_id,
mcp_tool_count=self._mcp_tool_count,
mcp_unauthenticated=self._mcp_unauthenticated,
@@ -2844,7 +2926,13 @@ class DeepAgentsApp(App):
gc.freeze()
chat = self.query_one("#chat", VerticalScroll)
chat.anchor()
# Don't establish bottom-follow intent at startup. `_ChatScroll.anchor()`
# defers the real anchor until content overflows, but not calling it at
# all keeps the welcome banner pinned to the top of an empty chat (like
# Claude Code). Content-producing paths (streaming, shell output,
# commands, model switches) call `anchor()` to opt into bottom-follow as
# content arrives; thread resume instead scrolls to the bottom once via
# `scroll_end()`.
self._apply_scrollbar_visibility(chat)
self._status_bar = self.query_one("#status-bar", StatusBar)
@@ -3878,6 +3966,10 @@ class DeepAgentsApp(App):
)
except NoMatches:
logger.warning("Welcome banner not found during server ready transition")
except ScreenStackError:
logger.debug(
"Screen stack empty during server ready transition", exc_info=True
)
self._sync_status_connection()
# Refresh the status bar model so a successful retry after a failed
@@ -3973,12 +4065,8 @@ class DeepAgentsApp(App):
)
logger.error("Server startup failed: %s", event.error, exc_info=event.error)
# Drop the banner's connecting spinner — chat surface owns the error.
try:
banner = self.query_one("#welcome-banner", WelcomeBanner)
banner.set_idle()
except NoMatches:
logger.warning("Welcome banner not found during server failure transition")
# The banner has no failure state — the status bar owns connection
# progress and the chat surface owns the error message below.
self._sync_status_connection()
# Keep any queued messages and widgets in place — `/model` retry can
@@ -9293,7 +9381,13 @@ class DeepAgentsApp(App):
banner = self.query_one("#welcome-banner", WelcomeBanner)
banner.update_thread_id(new_thread_id)
except NoMatches:
pass
# The banner is composed once and never removed, so a miss
# here means it has silently vanished — surface it.
logger.warning("Welcome banner not found during thread reset")
except ScreenStackError:
logger.debug(
"Screen stack empty during thread reset", exc_info=True
)
thread_msg_widget = AppMessage(f"Started new thread: {new_thread_id}")
await self._mount_message(thread_msg_widget)
self._schedule_thread_message_link(
@@ -10091,17 +10185,27 @@ class DeepAgentsApp(App):
return settings.model_provider or None
def _sync_status_model(self) -> None:
"""Update the status bar with the active model and reasoning effort."""
"""Update model displays with the active model and reasoning effort."""
from deepagents_code.config import settings
from deepagents_code.reasoning_effort import (
current_effort_from_model_params,
default_effort_for_model,
)
if self._status_bar is None:
return
provider = settings.model_provider or ""
model = settings.model_name or ""
try:
banner = self.query_one("#welcome-banner", WelcomeBanner)
banner.update_model(provider=provider, model=model)
except NoMatches:
# The banner is composed once and never removed, so a miss here in
# steady state means the live model row has silently stopped
# updating — surface it rather than swallow at debug.
logger.warning("Welcome banner not found during model sync")
except ScreenStackError:
logger.debug("Screen stack empty during model sync", exc_info=True)
if self._status_bar is None:
return
if not provider or not model:
logger.warning(
"Settings missing model identity at status sync "
@@ -13077,11 +13181,6 @@ class DeepAgentsApp(App):
self._connecting = True
self._reconnecting = True
self._agent = None
try:
banner = self.query_one("#welcome-banner", WelcomeBanner)
banner.set_connecting()
except NoMatches:
pass
self._sync_status_connection()
if self._chat_input:
@@ -13156,7 +13255,16 @@ class DeepAgentsApp(App):
mcp_awaiting_reconnect=self._mcp_awaiting_reconnect,
)
except NoMatches:
pass
# The banner is composed once and never removed, so a miss
# here means it has silently vanished — surface it.
logger.warning(
"Welcome banner not found during agent-swap rollback"
)
except ScreenStackError:
logger.debug(
"Screen stack empty during agent-swap rollback",
exc_info=True,
)
self._sync_status_connection()
self.notify(
f"Could not prepare to switch to {agent_name!r}. "
@@ -13217,7 +13325,16 @@ class DeepAgentsApp(App):
banner = self.query_one("#welcome-banner", WelcomeBanner)
banner.set_connected(self._mcp_tool_count)
except NoMatches:
pass
# The banner is composed once and never removed, so a miss here
# means it has silently vanished — surface it.
logger.warning(
"Welcome banner not found during agent-swap confirmation"
)
except ScreenStackError:
logger.debug(
"Screen stack empty during agent-swap confirmation",
exc_info=True,
)
self._sync_status_connection()
# Refresh skills so /skill: autocomplete reflects the new agent's
@@ -14005,10 +14122,15 @@ class DeepAgentsApp(App):
)
def _refresh_welcome_banner_mcp_counts(self) -> None:
"""Push current MCP counts into the welcome banner when it is mounted."""
"""Push current MCP counts into the welcome banner when it is mounted.
Best-effort: this can run during MCP initialization before the banner is
composed, so a miss is benign and stays at debug (unlike the post-startup
banner-update paths, where a miss signals a real regression).
"""
try:
banner = self.query_one("#welcome-banner", WelcomeBanner)
except NoMatches:
except (NoMatches, ScreenStackError):
logger.debug("Welcome banner not mounted during MCP count refresh")
return
banner.set_connected(
@@ -15116,11 +15238,6 @@ class DeepAgentsApp(App):
self._connecting = True
self._reconnecting = True
self._agent = None
try:
banner = self.query_one("#welcome-banner", WelcomeBanner)
banner.set_connecting()
except NoMatches:
pass
self._sync_status_connection()
try:
@@ -15251,6 +15368,8 @@ class DeepAgentsApp(App):
logger.warning(missing_message, thread_id)
else:
logger.debug(missing_message, thread_id)
except ScreenStackError:
logger.debug("Screen stack empty during thread-id sync", exc_info=True)
def _apply_cwd_to_ui(self, cwd: Path) -> None:
"""Update cwd-dependent UI state after changing process cwd."""
@@ -15260,6 +15379,14 @@ class DeepAgentsApp(App):
self._chat_input.set_cwd(cwd)
if self._status_bar is not None:
self._status_bar.cwd = cwd_text
try:
self.query_one("#welcome-banner", WelcomeBanner).update_cwd(cwd_text)
except NoMatches:
# Persistent banner: a steady-state miss means the live directory
# row has silently stopped updating — surface it.
logger.warning("Welcome banner not found during cwd sync")
except ScreenStackError:
logger.debug("Screen stack empty during cwd sync", exc_info=True)
@staticmethod
def _refresh_project_context_after_cwd_switch(cwd: Path) -> None:
@@ -15471,11 +15598,6 @@ class DeepAgentsApp(App):
self._connecting = True
self._reconnecting = True
self._agent = None
try:
banner = self.query_one("#welcome-banner", WelcomeBanner)
banner.set_connecting()
except NoMatches:
pass
self._sync_status_connection()
self._preserve_launch_relative_server_paths(previous_cwd)
self._switch_process_cwd(cwd)
@@ -15570,7 +15692,14 @@ class DeepAgentsApp(App):
mcp_awaiting_reconnect=self._mcp_awaiting_reconnect,
)
except NoMatches:
pass
# The banner is composed once and never removed, so a miss here
# means it has silently vanished — surface it.
logger.warning("Welcome banner not found during cwd-switch rollback")
except ScreenStackError:
logger.debug(
"Screen stack empty during cwd-switch rollback",
exc_info=True,
)
self._sync_status_connection()
if not isinstance(exc, Exception):
# Cancellation / SystemExit: state is restored; let it propagate.
@@ -16203,11 +16332,6 @@ class DeepAgentsApp(App):
self._server_startup_deferred = False
self._connecting = True
self._reconnecting = True
try:
banner = self.query_one("#welcome-banner", WelcomeBanner)
banner.set_connecting()
except (NoMatches, ScreenStackError):
logger.debug("Welcome banner not found during startup retry", exc_info=True)
self._sync_status_connection()
if self._retry_status_widget is not None:
libs/code/deepagents_code/config_manifest.py
+16 -15
View File
@@ -788,6 +788,22 @@ _STATIC_OPTIONS: tuple[ConfigOption, ...] = (
default=False,
env_var=_env_vars.SHOW_HEADER,
),
ConfigOption(
key="display.splash_show_model",
group="Display",
summary="Show the active model row in the startup welcome banner.",
kind=OptionKind.BOOL,
default=False,
env_var=_env_vars.SPLASH_SHOW_MODEL,
),
ConfigOption(
key="display.splash_show_cwd",
group="Display",
summary="Show the working-directory row in the startup welcome banner.",
kind=OptionKind.BOOL,
default=False,
env_var=_env_vars.SPLASH_SHOW_CWD,
),
ConfigOption(
key="display.kitty_keyboard",
group="Display",
@@ -836,14 +852,6 @@ _STATIC_OPTIONS: tuple[ConfigOption, ...] = (
default=True,
env_var=_env_vars.SHOW_LANGSMITH_REPLICA_TRACING,
),
ConfigOption(
key="display.hide_splash_tips",
group="Display",
summary="Hide rotating tips in the startup splash.",
kind=OptionKind.BOOL,
default=False,
env_var=_env_vars.HIDE_SPLASH_TIPS,
),
ConfigOption(
key="display.hide_splash_version",
group="Display",
@@ -1174,13 +1182,6 @@ _STATIC_OPTIONS: tuple[ConfigOption, ...] = (
default=False,
env_var=_env_vars.DEBUG_MCP_PROJECT_TRUST,
),
ConfigOption(
key="debug.override_startup_subheader",
group="Debug",
summary="Override the startup splash subheader text.",
kind=OptionKind.STR,
env_var=_env_vars.DANGEROUSLY_OVERRIDE_STARTUP_SUBHEADER,
),
)
libs/code/deepagents_code/widgets/welcome.py
+240 -269
View File
@@ -3,9 +3,9 @@
from __future__ import annotations
import asyncio
import os
import random
from typing import TYPE_CHECKING, Any
import logging
from pathlib import Path
from typing import TYPE_CHECKING, Any, Final
from textual.color import Color as TColor
from textual.content import Content
@@ -17,12 +17,13 @@ if TYPE_CHECKING:
from deepagents_code import theme
from deepagents_code._env_vars import (
DANGEROUSLY_OVERRIDE_STARTUP_SUBHEADER,
DEBUG,
HIDE_CWD,
HIDE_LANGSMITH_TRACING,
HIDE_SPLASH_TIPS,
HIDE_SPLASH_VERSION,
SHOW_LANGSMITH_REPLICA_TRACING,
SPLASH_SHOW_CWD,
SPLASH_SHOW_MODEL,
is_env_truthy,
)
from deepagents_code._version import __version__
@@ -30,14 +31,17 @@ from deepagents_code.config import (
_get_editable_install_path,
_is_editable_install,
fetch_langsmith_project_url,
get_banner,
get_glyphs,
get_langsmith_project_name,
get_langsmith_replica_project,
)
from deepagents_code.widgets._links import open_style_link
_LANGSMITH_UTM_SOURCE = "deepagents-code"
logger = logging.getLogger(__name__)
_ANSI_THEMES: Final[frozenset[str]] = frozenset({"ansi-dark", "ansi-light"})
"""Theme names whose color palette is determined by the terminal emulator
rather than by the app, so link styles use bold instead of a parsed color."""
_TIPS: dict[str, int] = {
"Use @ to reference files and / for commands": 3,
@@ -66,22 +70,10 @@ _TIPS: dict[str, int] = {
"Use !! for incognito shell commands that stay out of model context": 1,
"Deep Agents can explain its own features and look up its docs. Ask it how to use.": 3, # noqa: E501
}
"""Rotating tips shown in the welcome footer, with relative selection weights.
"""Rotating tips shown in the welcome footer, with relative selection weights."""
One is picked per session. Higher weights are picked more often.
"""
def _pick_tip() -> str:
"""Pick a tip from `_TIPS` weighted by its associated weight.
Returns:
A single tip string, selected with probability proportional to its
weight in `_TIPS`.
"""
tips = list(_TIPS.keys())
weights = list(_TIPS.values())
return random.choices(tips, weights=weights, k=1)[0] # noqa: S311
_LANGSMITH_UTM_SOURCE: Final[str] = "deepagents-code"
"""UTM source tag appended to LangSmith project URLs in the welcome banner."""
def _langsmith_project_link(project_url: str) -> str:
@@ -118,8 +110,40 @@ def _langsmith_project_link_style(
return TStyle(foreground=TColor.parse(colors.primary), link=link)
def _home_prefixed(cwd: str) -> str:
"""Format a directory path, using `~` for the home directory when possible.
Args:
cwd: Working directory path.
Returns:
The path with the home prefix collapsed to `~` when applicable.
"""
path = Path(cwd)
try:
home = Path.home()
if path == home:
return "~"
if path.is_relative_to(home):
return "~/" + path.relative_to(home).as_posix()
# `Path.home()` raises `RuntimeError` when the home dir can't be resolved
# (no HOME/passwd entry); `ValueError` guards odd paths (e.g. embedded NUL).
# Either way, fall back to the accurate absolute path.
except (ValueError, RuntimeError):
pass
return str(path)
class WelcomeBanner(Static):
"""Welcome banner displayed at startup."""
"""Compact welcome banner shown at startup.
Renders a bordered box with the product title and version, followed by rows
that appear only when their data (and any env gate) is present. In render
order: the active model (`SPLASH_SHOW_MODEL`, opt-in), working directory
(`SPLASH_SHOW_CWD`, opt-in), LangSmith tracing project and its replica (each
clickable once its URL resolves), thread ID (debug mode only), and the MCP
tool count. MCP server warnings and the editable-install path follow.
"""
# Disable Textual's auto_links to prevent a flicker cycle: Style.__add__
# calls .copy() for linked styles, generating a fresh random _link_id on
@@ -130,7 +154,8 @@ class WelcomeBanner(Static):
DEFAULT_CSS = """
WelcomeBanner {
height: auto;
padding: 1;
border: round $primary;
padding: 0 2;
margin-bottom: 1;
}
"""
@@ -140,6 +165,9 @@ class WelcomeBanner(Static):
thread_id: str | None = None,
mcp_tool_count: int = 0,
*,
model_provider: str = "",
model_name: str = "",
cwd: str | None = None,
mcp_unauthenticated: int = 0,
mcp_errored: int = 0,
mcp_awaiting_reconnect: int = 0,
@@ -148,24 +176,37 @@ class WelcomeBanner(Static):
"""Initialize the welcome banner.
Args:
thread_id: Optional thread ID to display in the banner.
thread_id: Displayed only when debug mode is enabled.
mcp_tool_count: Number of MCP tools loaded at startup.
model_provider: Active model provider (e.g. `anthropic`). Displayed
only when `SPLASH_SHOW_MODEL` is enabled.
model_name: Active model name. Displayed only when `SPLASH_SHOW_MODEL` is
enabled.
cwd: Working directory. Defaults to the process cwd. Displayed only
when `SPLASH_SHOW_CWD` is enabled.
mcp_unauthenticated: Number of MCP servers awaiting login.
mcp_errored: Number of MCP servers that failed to load.
mcp_awaiting_reconnect: Number of MCP servers that completed OAuth
login but are waiting for `/mcp reconnect` before their tools
can load.
mcp_awaiting_reconnect: Number of MCP servers awaiting reconnect.
**kwargs: Additional arguments passed to parent.
"""
self._model_provider = model_provider
self._model_name = model_name
self._cwd = cwd if cwd is not None else str(Path.cwd())
self._show_model = is_env_truthy(SPLASH_SHOW_MODEL)
# `_show_cwd` and `_hide_cwd` are deliberately orthogonal: `_show_cwd`
# gates the working-directory row (opt-in), while `_hide_cwd` gates the
# editable-install path below. They govern different surfaces, so both
# can be set without contradiction.
self._show_cwd = is_env_truthy(SPLASH_SHOW_CWD)
self._hide_cwd = is_env_truthy(HIDE_CWD)
self._hide_version = is_env_truthy(HIDE_SPLASH_VERSION)
# Avoid collision with Widget._thread_id (Textual internal int)
self._cli_thread_id: str | None = thread_id
self._cli_thread_id = thread_id
self._mcp_tool_count = mcp_tool_count
self._mcp_unauthenticated = mcp_unauthenticated
self._mcp_errored = mcp_errored
self._mcp_awaiting_reconnect = mcp_awaiting_reconnect
self._idle = False
self._hide_langsmith_tracing = is_env_truthy(HIDE_LANGSMITH_TRACING)
self._hide_splash_tips = is_env_truthy(HIDE_SPLASH_TIPS)
self._project_name: str | None = (
None if self._hide_langsmith_tracing else get_langsmith_project_name()
)
@@ -173,19 +214,17 @@ class WelcomeBanner(Static):
SHOW_LANGSMITH_REPLICA_TRACING,
default=True,
)
replica_project = (
self._replica_project: str | None = (
get_langsmith_replica_project()
if self._project_name and show_replica_tracing
else None
)
self._replica_projects: list[str] = [replica_project] if replica_project else []
self._project_urls: dict[str, str] = {}
self._tip: str | None = None if self._hide_splash_tips else _pick_tip()
self._show_thread_id = is_env_truthy(DEBUG)
super().__init__(self._build_banner(), **kwargs)
def on_mount(self) -> None:
"""Kick off background fetch for LangSmith project URL."""
"""Watch for theme changes and start the LangSmith project-URL fetch."""
self.watch(self.app, "theme", self._on_theme_change, init=False)
if self._project_name:
self.run_worker(self._fetch_and_update, exclusive=True)
@@ -198,9 +237,11 @@ class WelcomeBanner(Static):
"""Fetch the LangSmith URL in a thread and update the banner."""
if not self._project_name:
return
primary = self._project_name
project_urls: dict[str, str] = {}
projects = dict.fromkeys([primary, *self._replica_projects])
projects = dict.fromkeys(
project
for project in (self._project_name, self._replica_project)
if project
)
for project in projects:
try:
project_url = await asyncio.wait_for(
@@ -208,20 +249,58 @@ class WelcomeBanner(Static):
timeout=2.0,
)
except (TimeoutError, OSError):
logger.debug(
"LangSmith project URL fetch failed for %r", project, exc_info=True
)
project_url = None
if project_url:
project_urls[project] = project_url
self._project_urls = dict(project_urls)
self._project_urls[project] = project_url
self.update(self._build_banner())
def update_thread_id(self, thread_id: str) -> None:
"""Update the displayed thread ID and re-render the banner.
def _project_url(self, project: str | None) -> str | None:
"""Return the resolved LangSmith URL for a project.
Args:
thread_id: The new thread ID to display.
project: Project name to look up.
Returns:
Resolved project URL, or `None` when missing or not yet fetched.
"""
if project is None:
return None
return self._project_urls.get(project)
def update_model(self, *, provider: str, model: str) -> None:
"""Track a new model and re-render when it is displayed.
Args:
provider: Active model provider.
model: Active model name.
"""
self._model_provider = provider
self._model_name = model
if self._show_model:
self.update(self._build_banner())
def update_cwd(self, cwd: str) -> None:
"""Track a new working directory and re-render when it is displayed.
Args:
cwd: New working directory path.
"""
self._cwd = cwd
if self._show_cwd:
self.update(self._build_banner())
def update_thread_id(self, thread_id: str) -> None:
"""Track a new thread ID and re-render when debug mode is active.
Args:
thread_id: The new thread ID.
"""
self._cli_thread_id = thread_id
self.update(self._build_banner())
if self._show_thread_id:
self.update(self._build_banner())
def set_connected(
self,
@@ -231,293 +310,185 @@ class WelcomeBanner(Static):
mcp_errored: int = 0,
mcp_awaiting_reconnect: int = 0,
) -> None:
"""Render the ready banner footer after a successful connect.
The status bar owns visible connection progress; this just refreshes
the banner's tool counts and ready footer once the server is reachable.
"""Update MCP tool counts and re-render the banner.
Args:
mcp_tool_count: Number of MCP tools loaded during connection.
mcp_unauthenticated: Number of MCP servers awaiting login.
mcp_errored: Number of MCP servers that failed to load.
mcp_awaiting_reconnect: Number of MCP servers that completed OAuth
login but are waiting for `/mcp reconnect` before their tools
can load.
mcp_awaiting_reconnect: Number of MCP servers awaiting reconnect.
"""
self._idle = False
self._mcp_tool_count = mcp_tool_count
self._mcp_unauthenticated = mcp_unauthenticated
self._mcp_errored = mcp_errored
self._mcp_awaiting_reconnect = mcp_awaiting_reconnect
self.update(self._build_banner())
def set_connecting(self) -> None:
"""Render the regular banner footer during a reconnect.
The status bar owns visible connection progress. This method only
ensures the banner is no longer in the idle failure state.
"""
self._idle = False
self.update(self._build_banner())
def set_idle(self) -> None:
"""Transition to a neutral state with no footer.
Used after a fatal startup failure so the banner stops claiming
progress (the failure is communicated via the chat surface). The
banner keeps its identity rows (title, version, install path,
LangSmith project, thread ID) but appends no footer line, leaving
the chat error as the sole source of failure context.
"""
self._idle = True
self.update(self._build_banner())
def on_click(self, event: Click) -> None: # noqa: PLR6301 # Textual event handler
"""Open style-embedded hyperlinks on single click."""
open_style_link(event)
def on_mouse_move(self, event: MouseMove) -> None:
"""Show a hand pointer over link spans and reset it elsewhere.
`auto_links` is disabled to avoid a hover-refresh flicker, so the
pointer shape is updated manually from the style under the cursor.
"""
"""Show a hand pointer over link spans and reset it elsewhere."""
self.styles.pointer = "pointer" if event.style.link else "default"
def on_leave(self) -> None:
"""Reset the pointer shape when the mouse leaves the banner."""
self.styles.pointer = "default"
def _primary_project_url(
self,
project_urls: dict[str, str] | None = None,
) -> str | None:
"""Get the resolved LangSmith URL for the primary tracing project.
Args:
project_urls: Optional project URL mapping to use instead of cached
widget state.
Returns:
Primary project URL when resolved, otherwise `None`.
"""
if not self._project_name:
return None
urls = self._project_urls if project_urls is None else project_urls
return urls.get(self._project_name)
def _build_banner(
self,
project_urls: dict[str, str] | None = None,
) -> Content:
def _build_banner(self) -> Content:
"""Build the banner content.
When the primary project URL is resolved and a thread ID is set, the
thread ID is rendered as a clickable hyperlink to the LangSmith thread
view.
Args:
project_urls: LangSmith project URLs keyed by project name. Project
names with resolved URLs are rendered as links. When `None`,
cached widget state is used.
Returns:
Content object containing the formatted banner.
Content with the title and version, followed by any applicable rows
in order: model (when `SPLASH_SHOW_MODEL`), directory (when
`SPLASH_SHOW_CWD`), tracing and replica (each clickable once its URL
resolves), thread ID (debug only), MCP tool count, MCP server
warnings, and the editable-install path.
"""
parts: list[str | tuple[str, str | TStyle] | Content] = []
project_urls = self._project_urls if project_urls is None else project_urls
project_url = self._primary_project_url(project_urls)
colors = theme.get_theme_colors(self)
ansi = self.app.theme in {"ansi-dark", "ansi-light"}
banner = get_banner()
primary_style: str | TStyle = (
ansi = self.app.theme in _ANSI_THEMES
accent: str | TStyle = "bold" if ansi else colors.primary
title_style: str | TStyle = (
"bold"
if ansi
else TStyle(foreground=TColor.parse(colors.primary), bold=True)
)
warn_color: str = "bold yellow" if ansi else colors.warning
hide_version = is_env_truthy(HIDE_SPLASH_VERSION)
if not hide_version and not ansi and _is_editable_install():
# Highlight local-install version tag with tool accent; art stays primary.
dev_style = TStyle(foreground=TColor.parse(colors.tool), bold=True)
version_tag = f"v{__version__} (local)"
idx = banner.rfind(version_tag)
if idx >= 0:
parts.extend(
[
(banner[:idx], primary_style),
(version_tag, dev_style),
(banner[idx + len(version_tag) :] + "\n", primary_style),
]
)
else:
parts.append((banner + "\n", primary_style))
else:
parts.append((banner + "\n", primary_style))
# For ANSI theme, use "bold" (terminal foreground) instead of hex
accent: str | TStyle = "bold" if ansi else colors.primary
success_color: str = "bold green" if ansi else colors.success
hide_editable_path = hide_version or is_env_truthy(HIDE_CWD)
editable_path = None if hide_editable_path else _get_editable_install_path()
if editable_path:
parts.extend([("Installed from: ", "dim"), (editable_path, "dim"), "\n"])
if self._project_name:
parts.extend(
[
(f"{get_glyphs().checkmark} ", success_color),
"LangSmith tracing: ",
]
)
if project_url:
parts: list[str | tuple[str, str | TStyle]] = [
(f"{get_glyphs().cursor} ", title_style),
("dcode", "bold"),
]
if not self._hide_version:
parts.append((f" v{__version__}", "dim"))
if not ansi and _is_editable_install():
parts.append(
(
f"'{self._project_name}'",
_langsmith_project_link_style(
project_url,
ansi=ansi,
colors=colors,
),
" (local)",
TStyle(foreground=TColor.parse(colors.tool), bold=True),
)
)
else:
parts.append((f"'{self._project_name}'", accent))
parts.append("\n")
if self._replica_projects:
# `_replica_projects` holds at most one entry today (the server
# mirrors to a single extra project), but the loop renders any
# number so the splash stays correct if that limit is lifted.
parts.append((" Also tracing to: ", "dim"))
for idx, name in enumerate(self._replica_projects):
if idx:
parts.append((", ", "dim"))
parts.append(("'", "dim"))
replica_url = project_urls.get(name)
if replica_url:
parts.append(
(
name,
_langsmith_project_link_style(
replica_url,
ansi=ansi,
colors=colors,
),
)
)
else:
parts.append((name, "dim"))
parts.append(("'", "dim"))
parts.append("\n")
if self._cli_thread_id and not self._hide_langsmith_tracing:
# Row labels share a common column width so values stay aligned; the
# longest label ("directory:") needs 11 columns including its space.
rows: list[list[tuple[str, str | TStyle]]] = []
if self._show_model and self._model_name:
model_value = (
f"{self._model_provider}:{self._model_name}"
if self._model_provider
else self._model_name
)
rows.append([("model: ", "dim"), (model_value, accent)])
if self._show_cwd and self._cwd:
rows.append([("directory: ", "dim"), (_home_prefixed(self._cwd), accent)])
if self._project_name:
project_url = self._project_url(self._project_name)
if project_url:
thread_url = (
f"{project_url.rstrip('/')}/t/{self._cli_thread_id}"
"?utm_source=deepagents-code"
)
parts.extend(
rows.append(
[
(" Thread: ", "dim"),
(self._cli_thread_id, TStyle(dim=True, link=thread_url)),
("\n", "dim"),
("tracing: ", "dim"),
(
f"'{self._project_name}'",
_langsmith_project_link_style(
project_url,
ansi=ansi,
colors=colors,
),
),
]
)
else:
parts.append((f" Thread: {self._cli_thread_id}\n", "dim"))
rows.append(
[("tracing: ", "dim"), (f"'{self._project_name}'", accent)]
)
if self._replica_project:
replica_url = self._project_url(self._replica_project)
if replica_url:
rows.append(
[
("replica: ", "dim"),
(
f"'{self._replica_project}'",
_langsmith_project_link_style(
replica_url,
ansi=ansi,
colors=colors,
),
),
]
)
else:
rows.append(
[("replica: ", "dim"), (f"'{self._replica_project}'", accent)]
)
if self._show_thread_id and self._cli_thread_id:
rows.append([("thread: ", "dim"), (self._cli_thread_id, "dim")])
if self._mcp_tool_count > 0:
parts.append((f"{get_glyphs().checkmark} ", success_color))
label = "MCP tool" if self._mcp_tool_count == 1 else "MCP tools"
parts.append(f"Loaded {self._mcp_tool_count} {label}\n")
label = "tool" if self._mcp_tool_count == 1 else "tools"
rows.append(
[("mcp: ", "dim"), (f"{self._mcp_tool_count} {label}", accent)]
)
warn_color: str = "bold yellow" if ansi else colors.warning
for index, row in enumerate(rows):
parts.append("\n\n" if index == 0 else "\n")
parts.extend(row)
# MCP server warnings — actionable alerts not shown elsewhere in the banner.
warning_lines: list[list[tuple[str, str | TStyle]]] = []
if self._mcp_unauthenticated > 0:
server_label = "server" if self._mcp_unauthenticated == 1 else "servers"
verb = "needs" if self._mcp_unauthenticated == 1 else "need"
unauth_text = (
f"{self._mcp_unauthenticated} MCP {server_label} {verb} login "
"— open /mcp\n"
)
parts.extend(
warning_lines.append(
[
(f"{get_glyphs().warning} ", warn_color),
(unauth_text, "dim"),
(
(
f"{self._mcp_unauthenticated} MCP {server_label} {verb}"
" login — open /mcp"
),
"dim",
),
]
)
if self._mcp_errored > 0:
server_label = "server" if self._mcp_errored == 1 else "servers"
errored_text = (
f"{self._mcp_errored} MCP {server_label} failed to load "
"— open /mcp for details\n"
)
parts.extend(
warning_lines.append(
[
(f"{get_glyphs().warning} ", warn_color),
(errored_text, "dim"),
(
(
f"{self._mcp_errored} MCP {server_label} failed to load"
" — open /mcp for details"
),
"dim",
),
]
)
if self._mcp_awaiting_reconnect > 0:
server_label = "server" if self._mcp_awaiting_reconnect == 1 else "servers"
awaiting_text = (
f"{self._mcp_awaiting_reconnect} MCP {server_label} ready to load "
"— run `/mcp reconnect`\n"
)
parts.extend(
warning_lines.append(
[
(f"{get_glyphs().warning} ", warn_color),
(awaiting_text, "dim"),
(
(
f"{self._mcp_awaiting_reconnect} MCP {server_label}"
" ready to load — run `/mcp reconnect`"
),
"dim",
),
]
)
if not self._idle:
ready_color = "bold" if ansi else colors.primary
parts.append(
build_welcome_footer(
primary_color=ready_color,
tip=self._tip,
show_tip=not self._hide_splash_tips,
)
)
# `_idle` means no footer; chat-surface owns the failure message.
for line in warning_lines:
parts.append("\n")
parts.extend(line)
# Editable-install path for local development visibility.
if not self._hide_version and not self._hide_cwd:
editable_path = _get_editable_install_path()
if editable_path:
parts.append("\n")
parts.extend([("installed: ", "dim"), (editable_path, "dim")])
return Content.assemble(*parts)
def build_welcome_footer(
*,
primary_color: str = theme.PRIMARY,
tip: str | None = None,
show_tip: bool | None = None,
) -> Content:
"""Build the footer shown at the bottom of the welcome banner.
Includes a tip to help users discover features unless tips are disabled.
Args:
primary_color: Color string for the ready prompt.
Defaults to the module-level ANSI `PRIMARY` constant; widget callers
should pass the active theme's hex value.
tip: Tip text to display. When `None`, a random tip is selected.
Pass an explicit value to keep the tip stable across re-renders.
show_tip: Whether to show the tip. When `None`, the startup splash tips
env var controls visibility.
Returns:
Content with the ready prompt and, when enabled, a tip.
"""
if show_tip is None:
show_tip = not is_env_truthy(HIDE_SPLASH_TIPS)
if show_tip and tip is None:
tip = _pick_tip()
subheader = (
os.environ.get(DANGEROUSLY_OVERRIDE_STARTUP_SUBHEADER)
or "Ready to code! What would you like to build?"
)
parts: list[tuple[str, str]] = [(f"\n{subheader}", primary_color)]
if show_tip and tip is not None:
parts.append((f"\nTip: {tip}", "dim italic"))
return Content.assemble(*parts)
libs/code/tests/unit_tests/test_app.py
+315
View File
@@ -50,6 +50,7 @@ from deepagents_code.app import (
QueuedMessage,
TextualSessionState,
_build_whats_new_message,
_ChatScroll,
_display_model_label,
_extra_is_ready,
_parse_rubric_max_iterations,
@@ -19914,6 +19915,320 @@ class TestCanBypassQueue:
assert app._pending_messages[0].text == "/clear"
def _banner_query_raiser(app: DeepAgentsApp, exc: Exception) -> Callable[..., Widget]:
"""Return a `query_one` stand-in that raises for the welcome banner only.
Non-banner selectors delegate to the app's real `query_one`, so unrelated
lookups in the code under test keep working while the banner lookup fails.
Args:
app: App whose `query_one` is being replaced.
exc: Exception to raise when the welcome banner is queried.
Returns:
A callable suitable for `patch.object(app, "query_one", side_effect=...)`.
"""
real_query_one = app.query_one
def _query_one(selector: str, *args: object, **kwargs: object) -> Widget:
if selector == "#welcome-banner":
raise exc
return real_query_one(selector, *args, **kwargs)
return _query_one
class _ChatScrollTestApp(App[None]):
"""Small app for exercising chat scroll anchoring."""
CSS = """
#chat {
height: 5;
}
"""
def compose(self) -> ComposeResult:
"""Compose a short transcript in the chat scroll."""
with _ChatScroll(id="chat"):
yield Static("welcome", id="welcome")
yield Static("message", id="message")
class TestChatScrollAnchoring:
"""Regression coverage for welcome banner positioning."""
async def test_anchor_keeps_short_content_top_aligned(self) -> None:
"""Bottom-follow should not bottom-align content shorter than the viewport."""
app = _ChatScrollTestApp()
async with app.run_test() as pilot:
chat = app.query_one("#chat", _ChatScroll)
chat.anchor()
await pilot.pause()
assert chat.scroll_y == 0
assert not chat.is_anchored
chat.release_anchor()
chat.anchor()
await pilot.pause()
assert chat.scroll_y == 0
assert not chat.is_anchored
async def test_anchor_engages_after_content_overflows(self) -> None:
"""Deferred bottom-follow should engage once chat content overflows."""
app = _ChatScrollTestApp()
async with app.run_test() as pilot:
chat = app.query_one("#chat", _ChatScroll)
chat.anchor()
await pilot.pause()
await chat.mount(Static("\n".join(f"line {i}" for i in range(20))))
await pilot.pause()
await pilot.pause()
assert chat.is_anchored
assert chat.max_scroll_y > 0
assert chat.scroll_y == chat.max_scroll_y
async def test_anchor_engages_immediately_when_already_scrollable(self) -> None:
"""Resuming into an overflowing transcript anchors to the bottom at once."""
app = _ChatScrollTestApp()
async with app.run_test() as pilot:
chat = app.query_one("#chat", _ChatScroll)
# Content already overflows before we opt into bottom-follow (the
# thread-resume case), so anchor() must engage without deferral.
await chat.mount(Static("\n".join(f"line {i}" for i in range(20))))
await pilot.pause()
assert chat.max_scroll_y > 0
chat.anchor()
await pilot.pause()
await pilot.pause()
assert chat.is_anchored
assert chat.scroll_y == chat.max_scroll_y
async def test_anchor_resets_to_top_when_content_shrinks(self) -> None:
"""Content shrinking below the viewport releases the bottom anchor."""
app = _ChatScrollTestApp()
async with app.run_test() as pilot:
chat = app.query_one("#chat", _ChatScroll)
chat.anchor()
await pilot.pause()
tall = Static("\n".join(f"line {i}" for i in range(20)))
await chat.mount(tall)
await pilot.pause()
await pilot.pause()
assert chat.is_anchored
# Removing the overflow (e.g. /clear) must not leave the banner
# bottom-pinned in a near-empty chat.
await tall.remove()
await pilot.pause()
await pilot.pause()
assert not chat.is_anchored
assert chat.scroll_y == 0
async def test_release_anchor_stops_following_bottom(self) -> None:
"""After release_anchor, streamed content must not snap to the bottom."""
app = _ChatScrollTestApp()
async with app.run_test() as pilot:
chat = app.query_one("#chat", _ChatScroll)
chat.anchor()
await chat.mount(Static("\n".join(f"line {i}" for i in range(20))))
await pilot.pause()
await pilot.pause()
assert chat.is_anchored
assert chat.scroll_y == chat.max_scroll_y
# The user scrolls up: release bottom-follow and move off the end
# (Textual re-anchors automatically while parked at the bottom).
chat.release_anchor()
chat.scroll_to(y=0, animate=False, immediate=True)
await pilot.pause()
# More content streams in. Textual keeps `_anchored` set as standing
# intent after a release, so the behavioral signal is the scroll
# position: it must stay where the user left it, not snap to the new
# bottom.
await chat.mount(Static("\n".join(f"tail {i}" for i in range(20))))
await pilot.pause()
await pilot.pause()
assert chat.max_scroll_y > 0
assert chat.scroll_y == 0
async def test_anchor_reengages_after_release_when_scrollable(self) -> None:
"""A fresh anchor() after a manual scroll re-arms bottom-follow.
The app calls anchor() on every content-producing turn, so releasing
(user scrolls up) then re-anchoring must reset `_anchor_released` and
re-engage because content already overflows.
"""
app = _ChatScrollTestApp()
async with app.run_test() as pilot:
chat = app.query_one("#chat", _ChatScroll)
chat.anchor()
await chat.mount(Static("\n".join(f"line {i}" for i in range(20))))
await pilot.pause()
await pilot.pause()
assert chat.is_anchored
chat.release_anchor()
chat.scroll_to(y=0, animate=False, immediate=True)
await pilot.pause()
assert chat.scroll_y == 0
chat.anchor()
await pilot.pause()
await pilot.pause()
assert chat.is_anchored
assert chat.scroll_y == chat.max_scroll_y
async def test_anchor_false_disables_bottom_follow(self) -> None:
"""anchor(False) disarms bottom-follow and delegates to the base class."""
app = _ChatScrollTestApp()
async with app.run_test() as pilot:
chat = app.query_one("#chat", _ChatScroll)
chat.anchor()
await chat.mount(Static("\n".join(f"line {i}" for i in range(20))))
await pilot.pause()
await pilot.pause()
assert chat.is_anchored
chat.anchor(False)
await pilot.pause()
assert not chat._follow_bottom_when_scrollable
assert not chat.is_anchored
class TestWelcomeBannerLiveUpdates:
"""The banner starts top-aligned and mirrors live model/cwd changes.
These bind the PR's user-visible behaviors at the app<->widget seam: the
empty chat must not bottom-anchor at startup, and `/model` / cwd switches
must reach the banner (not just the status bar).
"""
async def test_startup_leaves_chat_top_aligned(self) -> None:
"""An empty chat keeps the welcome banner pinned to the top (unanchored)."""
app = DeepAgentsApp(agent=MagicMock(), thread_id="thread-123")
async with app.run_test() as pilot:
await pilot.pause()
chat = app.query_one("#chat", _ChatScroll)
assert chat.scroll_y == 0
assert not chat.is_anchored
async def test_sync_status_model_updates_banner(self) -> None:
"""`_sync_status_model` pushes the active model into the welcome banner."""
from deepagents_code._env_vars import SPLASH_SHOW_MODEL
from deepagents_code.widgets.welcome import WelcomeBanner
with patch.dict(os.environ, {SPLASH_SHOW_MODEL: "1"}):
app = DeepAgentsApp(agent=MagicMock(), thread_id="thread-123")
async with app.run_test() as pilot:
with patch("deepagents_code.config.settings") as mock_settings:
mock_settings.model_provider = "openai"
mock_settings.model_name = "gpt-5.5"
app._sync_status_model()
await pilot.pause()
banner = app.query_one("#welcome-banner", WelcomeBanner)
assert "openai:gpt-5.5" in banner._build_banner().plain
async def test_apply_cwd_updates_banner(self) -> None:
"""`_apply_cwd_to_ui` pushes the working directory into the welcome banner."""
from deepagents_code._env_vars import SPLASH_SHOW_CWD
from deepagents_code.widgets.welcome import WelcomeBanner
with patch.dict(os.environ, {SPLASH_SHOW_CWD: "1"}):
app = DeepAgentsApp(agent=MagicMock(), thread_id="thread-123")
async with app.run_test() as pilot:
app._apply_cwd_to_ui(Path("/work/proj"))
await pilot.pause()
banner = app.query_one("#welcome-banner", WelcomeBanner)
assert "/work/proj" in banner._build_banner().plain
async def test_sync_status_model_swallows_screen_stack_error(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""A `ScreenStackError` during banner lookup is absorbed at debug level."""
app = DeepAgentsApp(agent=MagicMock(), thread_id="thread-123")
async with app.run_test() as pilot:
await pilot.pause()
with (
patch("deepagents_code.config.settings") as mock_settings,
patch.object(
app,
"query_one",
side_effect=_banner_query_raiser(app, ScreenStackError("empty")),
),
caplog.at_level(logging.DEBUG, logger="deepagents_code.app"),
):
mock_settings.model_provider = "openai"
mock_settings.model_name = "gpt-5.5"
# Must not propagate — the guard exists precisely for this.
app._sync_status_model()
assert "Screen stack empty during model sync" in caplog.text
async def test_sync_status_model_warns_when_banner_missing(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""A `NoMatches` on the persistent banner is surfaced at warning level."""
app = DeepAgentsApp(agent=MagicMock(), thread_id="thread-123")
async with app.run_test() as pilot:
await pilot.pause()
with (
patch("deepagents_code.config.settings") as mock_settings,
patch.object(
app,
"query_one",
side_effect=_banner_query_raiser(app, NoMatches("gone")),
),
caplog.at_level(logging.WARNING, logger="deepagents_code.app"),
):
mock_settings.model_provider = "openai"
mock_settings.model_name = "gpt-5.5"
app._sync_status_model()
assert "Welcome banner not found during model sync" in caplog.text
async def test_apply_cwd_survives_missing_banner(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""`_apply_cwd_to_ui` still updates cwd state when the banner lookup fails."""
app = DeepAgentsApp(agent=MagicMock(), thread_id="thread-123")
async with app.run_test() as pilot:
await pilot.pause()
with (
patch.object(
app,
"query_one",
side_effect=_banner_query_raiser(app, NoMatches("gone")),
),
caplog.at_level(logging.WARNING, logger="deepagents_code.app"),
):
# Must not propagate; cwd state is set before the banner update.
app._apply_cwd_to_ui(Path("/work/proj"))
assert app._cwd == "/work/proj"
assert "Welcome banner not found during cwd sync" in caplog.text
class TestStatusBarConnectionMirroring:
"""The bottom status bar must mirror the connection + queue state.
libs/code/tests/unit_tests/test_welcome.py
+558 -737
View File
File diff suppressed because it is too large Load Diff