uvos/vibe
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

2.1.0 (#317)

Browse source 
Co-authored-by: Quentin Torroba <quentin.torroba@mistral.ai>
Co-authored-by: Michel Thomazo <michel.thomazo@mistral.ai>
Co-authored-by: Clément Drouin <clement.drouin@mistral.ai>
Co-authored-by: Vincent Guilloux <vincent.guilloux@mistral.ai>
Co-authored-by: Clément Siriex <clement.sirieix@mistral.ai>
Co-authored-by: Kim-Adeline Miguel <kimadeline.miguel@mistral.ai>
Co-authored-by: Nicolas Karolak <nicolas@karolak.fr>
This commit is contained in:
Mathias Gesbert 2026-02-11 18:17:30 +01:00 • committed by GitHub
parent 9809cfc831
commit 51fecc67d9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
176 changed files with 8652 additions and 4451 deletions
Download patch file Download diff file Expand all files Collapse all files

1
vibe/cli/cli.py
View file

@ -180,6 +180,7 @@ def run_cli(args: argparse.Namespace) -> None:
run_textual_ui(
agent_loop=agent_loop,
initial_prompt=args.initial_prompt or stdin_prompt,
teleport_on_start=args.teleport,
)
except (KeyboardInterrupt, EOFError):

67
vibe/cli/clipboard.py
View file

@ -1,13 +1,8 @@
from __future__ import annotations
import base64
from collections.abc import Callable
import os
import platform
import shutil
import subprocess
import pyperclip
from textual.app import App
_PREVIEW_MAX_LENGTH = 40
@ -24,33 +19,6 @@ def _copy_osc52(text: str) -> None:
tty.flush()
def _copy_x11_clipboard(text: str) -> None:
subprocess.run(
["xclip", "-selection", "clipboard"], input=text.encode("utf-8"), check=True
)
def _copy_wayland_clipboard(text: str) -> None:
subprocess.run(["wl-copy"], input=text.encode("utf-8"), check=True)
def _has_cmd(cmd: str) -> bool:
return shutil.which(cmd) is not None
def _get_copy_fns(app: App) -> list[Callable[[str], None]]:
copy_fns: list[Callable[[str], None]] = [
_copy_osc52,
pyperclip.copy,
app.copy_to_clipboard,
]
if platform.system() == "Linux" and _has_cmd("wl-copy"):
copy_fns = [_copy_wayland_clipboard, *copy_fns]
if platform.system() == "Linux" and _has_cmd("xclip"):
copy_fns = [_copy_x11_clipboard, *copy_fns]
return copy_fns
def _shorten_preview(texts: list[str]) -> str:
dense_text = "⏎".join(texts).replace("\n", "⏎")
if len(dense_text) > _PREVIEW_MAX_LENGTH:
@ -58,7 +26,7 @@ def _shorten_preview(texts: list[str]) -> str:
return dense_text
def copy_selection_to_clipboard(app: App) -> None:
def copy_selection_to_clipboard(app: App, show_toast: bool = True) -> None:
selected_texts = []
for widget in app.query("*"):
@ -84,27 +52,16 @@ def copy_selection_to_clipboard(app: App) -> None:
combined_text = "\n".join(selected_texts)
success = False
copy_fns = _get_copy_fns(app)
for copy_fn in copy_fns:
try:
copy_fn(combined_text)
except:
pass
else:
success = True
if success:
try:
_copy_osc52(combined_text)
if show_toast:
app.notify(
f'"{_shorten_preview(selected_texts)}" copied to clipboard',
severity="information",
timeout=2,
markup=False,
)
except Exception:
app.notify(
f'"{_shorten_preview(selected_texts)}" copied to clipboard',
severity="information",
timeout=2,
markup=False,
)
else:
app.notify(
"Failed to copy - no clipboard method available",
severity="warning",
timeout=3,
"Failed to copy - clipboard not available", severity="warning", timeout=3
)

8
vibe/cli/commands.py
View file

@ -22,7 +22,7 @@ class CommandRegistry:
handler="_show_help",
),
"config": Command(
aliases=frozenset(["/config", "/theme", "/model"]),
aliases=frozenset(["/config", "/model"]),
description="Edit config settings",
handler="_show_config",
),
@ -62,6 +62,11 @@ class CommandRegistry:
description="Display agent statistics",
handler="_show_status",
),
"teleport": Command(
aliases=frozenset(["/teleport"]),
description="Teleport session to Vibe Nuage",
handler="_teleport_command",
),
}
for command in excluded_commands:
@ -86,7 +91,6 @@ class CommandRegistry:
"- `Ctrl+C` Quit (or clear input if text present)",
"- `Ctrl+G` Edit input in external editor",
"- `Ctrl+O` Toggle tool output view",
"- `Ctrl+T` Toggle todo view",
"- `Shift+Tab` Toggle auto-approve mode",
"",
"### Special Features",

3
vibe/cli/entrypoint.py
View file

@ -84,6 +84,9 @@ def parse_arguments() -> argparse.Namespace:
help="Change to this directory before running",
)
# Feature flag for teleport, not exposed to the user yet
parser.add_argument("--teleport", action="store_true", help=argparse.SUPPRESS)
continuation_group = parser.add_mutually_exclusive_group()
continuation_group.add_argument(
"-c",

23
vibe/cli/plan_offer/decide_plan_offer.py
View file

@ -2,6 +2,7 @@ from __future__ import annotations
from enum import StrEnum
import logging
from os import getenv
from vibe.cli.plan_offer.ports.whoami_gateway import (
WhoAmIGateway,
@ -9,6 +10,7 @@ from vibe.cli.plan_offer.ports.whoami_gateway import (
WhoAmIGatewayUnauthorized,
WhoAmIResponse,
)
from vibe.core.config import DEFAULT_MISTRAL_API_ENV_KEY, Backend, ProviderConfig
logger = logging.getLogger(__name__)
@ -62,3 +64,24 @@ def _action_and_plan_from_response(
return PlanOfferAction.UPGRADE, PlanType.FREE
case _:
return PlanOfferAction.NONE, PlanType.UNKNOWN
def resolve_api_key_for_plan(provider: ProviderConfig) -> str | None:
api_env_key = DEFAULT_MISTRAL_API_ENV_KEY
if provider.backend == Backend.MISTRAL:
api_env_key = provider.api_key_env_var
return getenv(api_env_key)
def plan_offer_cta(action: PlanOfferAction) -> str | None:
if action is PlanOfferAction.NONE:
return
url = ACTION_TO_URL[action]
match action:
case PlanOfferAction.UPGRADE:
text = f"### Unlock more with Vibe - [Upgrade to Le Chat Pro]({url})"
case PlanOfferAction.SWITCH_TO_PRO_KEY:
text = f"### Switch to your [Le Chat Pro API key]({url})"
return text

128
vibe/cli/terminal_setup.py
View file

@ -272,117 +272,29 @@ def _setup_iterm2() -> SetupResult:
def _setup_wezterm() -> SetupResult:
wezterm_config = Path.home() / ".wezterm.lua"
key_binding = """{
key = "Enter",
mods = "SHIFT",
action = wezterm.action.SendString("\\x1b[13;2u"),
}"""
try:
if wezterm_config.exists():
content = wezterm_config.read_text()
if 'mods = "SHIFT"' in content and 'key = "Enter"' in content:
return SetupResult(
success=True,
terminal=Terminal.WEZTERM,
message="Shift+Enter already configured in WezTerm",
)
if "keys = {" in content:
content = content.replace("keys = {", f"keys = {{\n {key_binding},")
else:
return SetupResult(
success=False,
terminal=Terminal.WEZTERM,
message="Please manually add the following to your .wezterm.lua:\n\n"
f" keys = {{\n {key_binding}\n }}",
)
else:
content = f"""local wezterm = require 'wezterm'
return {{
keys = {{
{key_binding}
}},
}}
"""
wezterm_config.write_text(content)
return SetupResult(
success=True,
terminal=Terminal.WEZTERM,
message=f"Added Shift+Enter binding to {wezterm_config}",
requires_restart=True,
)
except Exception as e:
return SetupResult(
success=False,
terminal=Terminal.WEZTERM,
message=f"Failed to configure WezTerm: {e}",
)
return SetupResult(
success=True,
terminal=Terminal.WEZTERM,
message="Please manually add the following to your .wezterm.lua:\n"
"local wezterm = require 'wezterm'\n"
"local config = wezterm.config_builder()\n\n"
"config.keys = {\n"
" {\n"
' key = "Enter",\n'
' mods = "SHIFT",\n'
' action = wezterm.action.SendString("\\x1b[13;2u"),\n'
" }\n"
"}\n\n"
"return config",
)
def _setup_ghostty() -> SetupResult:
system = platform.system()
if system == "Darwin":
config_path = (
Path.home()
/ "Library"
/ "Application Support"
/ "com.mitchellh.ghostty"
/ "config"
)
elif system == "Linux":
xdg_config = os.environ.get("XDG_CONFIG_HOME", str(Path.home() / ".config"))
config_path = Path(xdg_config) / "ghostty" / "config"
else:
return SetupResult(
success=False,
terminal=Terminal.GHOSTTY,
message="Ghostty configuration path unknown for this OS",
)
keybind_line = "keybind = shift+enter=text:\\x1b[13;2u"
try:
if config_path.exists():
content = config_path.read_text()
if "shift+enter" in content.lower():
return SetupResult(
success=True,
terminal=Terminal.GHOSTTY,
message="Shift+Enter already configured in Ghostty",
)
if not content.endswith("\n"):
content += "\n"
content += keybind_line + "\n"
else:
config_path.parent.mkdir(parents=True, exist_ok=True)
content = keybind_line + "\n"
config_path.write_text(content)
return SetupResult(
success=True,
terminal=Terminal.GHOSTTY,
message=f"Added Shift+Enter binding to {config_path}",
requires_restart=True,
)
except Exception as e:
return SetupResult(
success=False,
terminal=Terminal.GHOSTTY,
message=f"Failed to configure Ghostty: {e}",
)
return SetupResult(
success=True,
terminal=Terminal.GHOSTTY,
message="Shift+Enter is already configured in Ghostty",
)
def setup_terminal() -> SetupResult:

58
vibe/cli/textual_ui/ansi_markdown.py Normal file
View file

@ -0,0 +1,58 @@
from __future__ import annotations
from pygments.token import Token
from textual.content import Content
from textual.highlight import HighlightTheme, highlight
from textual.widgets import Markdown
from textual.widgets._markdown import MarkdownFence
class AnsiHighlightTheme(HighlightTheme):
STYLES = {
Token.Comment: "ansi_bright_black italic",
Token.Error: "ansi_red",
Token.Generic.Strong: "bold",
Token.Generic.Emph: "italic",
Token.Generic.Error: "ansi_red",
Token.Generic.Heading: "ansi_blue underline",
Token.Generic.Subheading: "ansi_blue",
Token.Keyword: "ansi_magenta",
Token.Keyword.Constant: "ansi_cyan",
Token.Keyword.Namespace: "ansi_magenta",
Token.Keyword.Type: "ansi_cyan",
Token.Literal.Number: "ansi_yellow",
Token.Literal.String.Backtick: "ansi_bright_black",
Token.Literal.String: "ansi_green",
Token.Literal.String.Doc: "ansi_green italic",
Token.Literal.String.Double: "ansi_green",
Token.Name: "ansi_default",
Token.Name.Attribute: "ansi_yellow",
Token.Name.Builtin: "ansi_cyan",
Token.Name.Builtin.Pseudo: "italic",
Token.Name.Class: "ansi_yellow",
Token.Name.Constant: "ansi_red",
Token.Name.Decorator: "ansi_blue",
Token.Name.Function: "ansi_blue",
Token.Name.Function.Magic: "ansi_blue",
Token.Name.Tag: "ansi_blue",
Token.Name.Variable: "ansi_default",
Token.Number: "ansi_yellow",
Token.Operator: "ansi_default",
Token.Operator.Word: "ansi_magenta",
Token.String: "ansi_green",
Token.Whitespace: "",
}
class AnsiMarkdownFence(MarkdownFence):
@classmethod
def highlight(cls, code: str, language: str) -> Content:
return highlight(code, language=language or None, theme=AnsiHighlightTheme)
class AnsiMarkdown(Markdown):
BLOCKS = {
**Markdown.BLOCKS,
"fence": AnsiMarkdownFence,
"code_block": AnsiMarkdownFence,
}

583
vibe/cli/textual_ui/app.py
View file

@ -2,16 +2,16 @@ from __future__ import annotations
import asyncio
from enum import StrEnum, auto
from os import getenv
from pathlib import Path
import subprocess
import time
from typing import Any, ClassVar, assert_never, cast
from weakref import WeakKeyDictionary
from pydantic import BaseModel
from textual.app import App, ComposeResult
from textual.binding import Binding, BindingType
from textual.containers import Horizontal, VerticalScroll
from textual.containers import Horizontal, VerticalGroup, VerticalScroll
from textual.events import AppBlur, AppFocus, MouseUp
from textual.widget import Widget
from textual.widgets import Static
@ -21,31 +21,27 @@ from vibe.cli.clipboard import copy_selection_to_clipboard
from vibe.cli.commands import CommandRegistry
from vibe.cli.plan_offer.adapters.http_whoami_gateway import HttpWhoAmIGateway
from vibe.cli.plan_offer.decide_plan_offer import (
ACTION_TO_URL,
PlanOfferAction,
PlanType,
decide_plan_offer,
plan_offer_cta,
resolve_api_key_for_plan,
)
from vibe.cli.plan_offer.ports.whoami_gateway import WhoAmIGateway
from vibe.cli.terminal_setup import setup_terminal
from vibe.cli.textual_ui.handlers.event_handler import EventHandler
from vibe.cli.textual_ui.terminal_theme import (
TERMINAL_THEME_NAME,
capture_terminal_theme,
)
from vibe.cli.textual_ui.widgets.agent_indicator import AgentIndicator
from vibe.cli.textual_ui.widgets.approval_app import ApprovalApp
from vibe.cli.textual_ui.widgets.banner.banner import Banner
from vibe.cli.textual_ui.widgets.chat_input import ChatInputContainer
from vibe.cli.textual_ui.widgets.compact import CompactMessage
from vibe.cli.textual_ui.widgets.config_app import ConfigApp
from vibe.cli.textual_ui.widgets.context_progress import ContextProgress, TokenState
from vibe.cli.textual_ui.widgets.load_more import HistoryLoadMoreRequested
from vibe.cli.textual_ui.widgets.loading import LoadingWidget, paused_timer
from vibe.cli.textual_ui.widgets.messages import (
AssistantMessage,
BashOutputMessage,
ErrorMessage,
InterruptMessage,
PlanOfferMessage,
ReasoningMessage,
StreamingMessageBase,
UserCommandMessage,
@ -56,8 +52,19 @@ from vibe.cli.textual_ui.widgets.messages import (
from vibe.cli.textual_ui.widgets.no_markup_static import NoMarkupStatic
from vibe.cli.textual_ui.widgets.path_display import PathDisplay
from vibe.cli.textual_ui.widgets.question_app import QuestionApp
from vibe.cli.textual_ui.widgets.teleport_message import TeleportMessage
from vibe.cli.textual_ui.widgets.tools import ToolCallMessage, ToolResultMessage
from vibe.cli.textual_ui.widgets.welcome import WelcomeBanner
from vibe.cli.textual_ui.windowing import (
HISTORY_RESUME_TAIL_MESSAGES,
LOAD_MORE_BATCH_SIZE,
HistoryLoadMoreManager,
SessionWindowing,
build_history_widgets,
create_resume_plan,
non_system_history_messages,
should_resume_history,
sync_backfill_state,
)
from vibe.cli.update_notifier import (
FileSystemUpdateCacheRepository,
PyPIUpdateGateway,
@ -70,16 +77,29 @@ from vibe.cli.update_notifier import (
should_show_whats_new,
)
from vibe.cli.update_notifier.update import do_update
from vibe.core.agent_loop import AgentLoop
from vibe.core.agent_loop import AgentLoop, TeleportError
from vibe.core.agents import AgentProfile
from vibe.core.autocompletion.path_prompt_adapter import render_path_prompt
from vibe.core.config import VibeConfig
from vibe.core.paths.config_paths import HISTORY_FILE
from vibe.core.session.session_loader import SessionLoader
from vibe.core.teleport.types import (
TeleportAuthCompleteEvent,
TeleportAuthRequiredEvent,
TeleportCheckingGitEvent,
TeleportCompleteEvent,
TeleportPushingEvent,
TeleportPushRequiredEvent,
TeleportPushResponseEvent,
TeleportSendingGithubTokenEvent,
TeleportStartingWorkflowEvent,
)
from vibe.core.tools.base import ToolPermission
from vibe.core.tools.builtins.ask_user_question import (
AskUserQuestionArgs,
AskUserQuestionResult,
Choice,
Question,
)
from vibe.core.types import (
AgentStats,
@ -110,6 +130,47 @@ class BottomApp(StrEnum):
Question = auto()
class ChatScroll(VerticalScroll):
"""Optimized scroll container that skips cascading style recalculations."""
def update_node_styles(self, animate: bool = True) -> None:
pass
PRUNE_LOW_MARK = 1000
PRUNE_HIGH_MARK = 1500
async def prune_by_height(messages_area: Widget, low_mark: int, high_mark: int) -> bool:
"""Remove older children to keep virtual height within bounds.
Implementation from https://github.com/batrachianai/toad/blob/a335b56c9015514d5f38654e3909aaa78850c510/src/toad/widgets/conversation.py#L1495
"""
height = messages_area.virtual_size.height
if height <= high_mark:
return False
prune_children: list[Widget] = []
bottom_margin = 0
prune_height = 0
for child in messages_area.children:
if not child.display:
prune_children.append(child)
continue
top, _, bottom, _ = child.styles.margin
child_height = child.outer_size.height
prune_height = (
(prune_height - bottom_margin + max(bottom_margin, top))
+ bottom
+ child_height
)
bottom_margin = bottom
if height - prune_height <= low_mark:
break
prune_children.append(child)
if prune_children:
await messages_area.remove_children(prune_children)
return bool(prune_children)
class VibeApp(App): # noqa: PLR0904
ENABLE_COMMAND_PALETTE = False
CSS_PATH = "app.tcss"
@ -119,7 +180,8 @@ class VibeApp(App): # noqa: PLR0904
Binding("ctrl+d", "force_quit", "Quit", show=False, priority=True),
Binding("escape", "interrupt", "Interrupt", show=False, priority=True),
Binding("ctrl+o", "toggle_tool", "Toggle Tool", show=False),
Binding("ctrl+t", "toggle_todo", "Toggle Todo", show=False),
Binding("ctrl+y", "copy_selection", "Copy", show=False, priority=True),
Binding("ctrl+shift+c", "copy_selection", "Copy", show=False, priority=True),
Binding("shift+tab", "cycle_mode", "Cycle Mode", show=False, priority=True),
Binding("shift+up", "scroll_chat_up", "Scroll Up", show=False, priority=True),
Binding(
@ -131,6 +193,7 @@ class VibeApp(App): # noqa: PLR0904
self,
agent_loop: AgentLoop,
initial_prompt: str | None = None,
teleport_on_start: bool = False,
update_notifier: UpdateGateway | None = None,
update_cache_repository: UpdateCacheRepository | None = None,
current_version: str = CORE_VERSION,
@ -148,42 +211,51 @@ class VibeApp(App): # noqa: PLR0904
self._pending_question: asyncio.Future | None = None
self.event_handler: EventHandler | None = None
self.commands = CommandRegistry()
excluded_commands = []
if not self.config.nuage_enabled:
excluded_commands.append("teleport")
self.commands = CommandRegistry(excluded_commands=excluded_commands)
self._chat_input_container: ChatInputContainer | None = None
self._agent_indicator: AgentIndicator | None = None
self._current_bottom_app: BottomApp = BottomApp.Input
self.history_file = HISTORY_FILE.path
self._tools_collapsed = True
self._todos_collapsed = False
self._current_streaming_message: AssistantMessage | None = None
self._current_streaming_reasoning: ReasoningMessage | None = None
self._windowing = SessionWindowing(load_more_batch_size=LOAD_MORE_BATCH_SIZE)
self._load_more = HistoryLoadMoreManager()
self._tool_call_map: dict[str, str] | None = None
self._history_widget_indices: WeakKeyDictionary[Widget, int] = (
WeakKeyDictionary()
)
self._update_notifier = update_notifier
self._update_cache_repository = update_cache_repository
self._current_version = current_version
self._plan_offer_gateway = plan_offer_gateway
self._plan_offer_shown = False
self._initial_prompt = initial_prompt
self._teleport_on_start = teleport_on_start and self.config.nuage_enabled
self._auto_scroll = True
self._last_escape_time: float | None = None
self._terminal_theme = capture_terminal_theme()
self._banner: Banner | None = None
self._cached_messages_area: Widget | None = None
self._cached_chat: ChatScroll | None = None
self._cached_loading_area: Widget | None = None
@property
def config(self) -> VibeConfig:
return self.agent_loop.config
def compose(self) -> ComposeResult:
with VerticalScroll(id="chat"):
yield WelcomeBanner(self.config)
yield Static(id="messages")
with ChatScroll(id="chat"):
self._banner = Banner(self.config, self.agent_loop.skill_manager)
yield self._banner
yield VerticalGroup(id="messages")
with Horizontal(id="loading-area"):
yield Static(id="loading-area-content")
yield AgentIndicator(profile=self.agent_loop.agent_profile)
yield Static(id="todo-area")
with Static(id="bottom-app-container"):
yield ChatInputContainer(
@ -191,7 +263,9 @@ class VibeApp(App): # noqa: PLR0904
command_registry=self.commands,
id="input-container",
safety=self.agent_loop.agent_profile.safety,
agent_name=self.agent_loop.agent_profile.display_name.lower(),
skill_entries_getter=self._get_skill_entries,
nuage_enabled=self.config.nuage_enabled,
)
with Horizontal(id="bottom-bar"):
@ -200,25 +274,19 @@ class VibeApp(App): # noqa: PLR0904
yield ContextProgress()
async def on_mount(self) -> None:
if self._terminal_theme:
self.register_theme(self._terminal_theme)
self.theme = "textual-ansi"
if self.config.textual_theme == TERMINAL_THEME_NAME:
if self._terminal_theme:
self.theme = TERMINAL_THEME_NAME
else:
self.theme = self.config.textual_theme
self._cached_messages_area = self.query_one("#messages")
self._cached_chat = self.query_one("#chat", ChatScroll)
self._cached_loading_area = self.query_one("#loading-area-content")
self.event_handler = EventHandler(
mount_callback=self._mount_and_scroll,
scroll_callback=self._scroll_to_bottom_deferred,
todo_area_callback=lambda: self.query_one("#todo-area"),
get_tools_collapsed=lambda: self._tools_collapsed,
get_todos_collapsed=lambda: self._todos_collapsed,
)
self._chat_input_container = self.query_one(ChatInputContainer)
self._agent_indicator = self.query_one(AgentIndicator)
context_progress = self.query_one(ContextProgress)
def update_context_progress(stats: AgentStats) -> None:
@ -227,7 +295,7 @@ class VibeApp(App): # noqa: PLR0904
current_tokens=stats.context_tokens,
)
AgentStats.add_listener("context_tokens", update_context_progress)
self.agent_loop.stats.add_listener("context_tokens", update_context_progress)
self.agent_loop.stats.trigger_listeners()
self.agent_loop.set_approval_callback(self._approval_callback)
@ -237,17 +305,19 @@ class VibeApp(App): # noqa: PLR0904
chat_input_container = self.query_one(ChatInputContainer)
chat_input_container.focus_input()
await self._show_dangerous_directory_warning()
await self._resume_history_from_messages()
await self._check_and_show_whats_new()
await self._maybe_show_plan_offer()
self._schedule_update_notification()
await self._rebuild_history_from_messages()
if self._initial_prompt:
if self._initial_prompt or self._teleport_on_start:
self.call_after_refresh(self._process_initial_prompt)
def _process_initial_prompt(self) -> None:
if self._initial_prompt:
if self._teleport_on_start:
self.run_worker(
self._handle_teleport_command(self._initial_prompt), exclusive=False
)
elif self._initial_prompt:
self.run_worker(
self._handle_user_message(self._initial_prompt), exclusive=False
)
@ -255,6 +325,9 @@ class VibeApp(App): # noqa: PLR0904
async def on_chat_input_container_submitted(
self, event: ChatInputContainer.Submitted
) -> None:
if self._banner:
self._banner.freeze_animation()
value = event.value.strip()
if not value:
return
@ -269,6 +342,11 @@ class VibeApp(App): # noqa: PLR0904
await self._handle_bash_command(value[1:])
return
if value.startswith("&"):
if self.config.nuage_enabled:
await self._handle_teleport_command(value[1:])
return
if await self._handle_command(value):
return
@ -330,29 +408,6 @@ class VibeApp(App): # noqa: PLR0904
if self._loading_widget and self._loading_widget.parent:
await self._loading_widget.remove()
self._loading_widget = None
self._hide_todo_area()
def _show_todo_area(self) -> None:
try:
todo_area = self.query_one("#todo-area")
todo_area.add_class("loading-active")
except Exception:
pass
def _hide_todo_area(self) -> None:
try:
todo_area = self.query_one("#todo-area")
todo_area.remove_class("loading-active")
except Exception:
pass
def on_config_app_setting_changed(self, message: ConfigApp.SettingChanged) -> None:
if message.key == "textual_theme":
if message.value == TERMINAL_THEME_NAME:
if self._terminal_theme:
self.theme = TERMINAL_THEME_NAME
else:
self.theme = message.value
async def on_config_app_config_closed(
self, message: ConfigApp.ConfigClosed
@ -370,7 +425,7 @@ class VibeApp(App): # noqa: PLR0904
async def on_compact_message_completed(
self, message: CompactMessage.Completed
) -> None:
messages_area = self.query_one("#messages")
messages_area = self._cached_messages_area or self.query_one("#messages")
children = list(messages_area.children)
try:
@ -483,62 +538,62 @@ class VibeApp(App): # noqa: PLR0904
self._handle_agent_loop_turn(message)
)
async def _rebuild_history_from_messages(self) -> None:
if all(msg.role == Role.system for msg in self.agent_loop.messages):
async def _resume_history_from_messages(self) -> None:
messages_area = self._cached_messages_area or self.query_one("#messages")
if not should_resume_history(list(messages_area.children)):
return
messages_area = self.query_one("#messages")
# Don't rebuild if messages are already displayed
if messages_area.children:
self._windowing.reset()
history_messages = non_system_history_messages(self.agent_loop.messages)
if (
plan := create_resume_plan(history_messages, HISTORY_RESUME_TAIL_MESSAGES)
) is None:
return
await self._mount_history_batch(
plan.tail_messages,
messages_area,
plan.tool_call_map,
start_index=plan.tail_start_index,
)
self.call_after_refresh(
lambda: self._align_chat_after_history_rebuild(plan.has_backfill)
)
self._tool_call_map = plan.tool_call_map
self._windowing.set_backfill(plan.backfill_messages)
await self._load_more.set_visible(
messages_area,
visible=self._windowing.has_backfill,
remaining=self._windowing.remaining,
)
tool_call_map: dict[str, str] = {}
async def _mount_history_batch(
self,
batch: list[LLMMessage],
messages_area: Widget,
tool_call_map: dict[str, str],
*,
start_index: int,
before: Widget | int | None = None,
after: Widget | None = None,
) -> None:
widgets = build_history_widgets(
batch=batch,
tool_call_map=tool_call_map,
start_index=start_index,
tools_collapsed=self._tools_collapsed,
history_widget_indices=self._history_widget_indices,
)
with self.batch_update():
for msg in self.agent_loop.messages:
if msg.role == Role.system:
continue
match msg.role:
case Role.user:
if msg.content:
await messages_area.mount(UserMessage(msg.content))
case Role.assistant:
await self._mount_history_assistant_message(
msg, messages_area, tool_call_map
)
case Role.tool:
tool_name = msg.name or tool_call_map.get(
msg.tool_call_id or "", "tool"
)
await messages_area.mount(
ToolResultMessage(
tool_name=tool_name,
content=msg.content,
collapsed=self._tools_collapsed,
)
)
async def _mount_history_assistant_message(
self, msg: LLMMessage, messages_area: Widget, tool_call_map: dict[str, str]
) -> None:
if msg.content:
widget = AssistantMessage(msg.content)
await messages_area.mount(widget)
await widget.write_initial_content()
await widget.stop_stream()
if not msg.tool_calls:
return
for tool_call in msg.tool_calls:
tool_name = tool_call.function.name or "unknown"
if tool_call.id:
tool_call_map[tool_call.id] = tool_name
await messages_area.mount(ToolCallMessage(tool_name=tool_name))
if not widgets:
return
if before is not None:
await messages_area.mount_all(widgets, before=before)
return
if after is not None:
await messages_area.mount_all(widgets, after=after)
return
await messages_area.mount_all(widgets)
def _is_tool_enabled_in_main_agent(self, tool: str) -> bool:
return tool in self.agent_loop.tool_manager.available_tools
@ -574,12 +629,13 @@ class VibeApp(App): # noqa: PLR0904
async def _handle_agent_loop_turn(self, prompt: str) -> None:
self._agent_running = True
loading_area = self.query_one("#loading-area-content")
loading_area = self._cached_loading_area or self.query_one(
"#loading-area-content"
)
loading = LoadingWidget()
self._loading_widget = loading
await loading_area.mount(loading)
self._show_todo_area()
try:
rendered_prompt = render_path_prompt(prompt, base_dir=Path.cwd())
@ -595,13 +651,13 @@ class VibeApp(App): # noqa: PLR0904
if self._loading_widget and self._loading_widget.parent:
await self._loading_widget.remove()
if self.event_handler:
self.event_handler.stop_current_tool_call()
self.event_handler.stop_current_tool_call(success=False)
raise
except Exception as e:
if self._loading_widget and self._loading_widget.parent:
await self._loading_widget.remove()
if self.event_handler:
self.event_handler.stop_current_tool_call()
self.event_handler.stop_current_tool_call(success=False)
message = str(e)
if isinstance(e, RateLimitError):
@ -620,8 +676,100 @@ class VibeApp(App): # noqa: PLR0904
if self._loading_widget:
await self._loading_widget.remove()
self._loading_widget = None
self._hide_todo_area()
await self._finalize_current_streaming_message()
await self._refresh_windowing_from_history()
async def _teleport_command(self) -> None:
await self._handle_teleport_command(show_message=False)
async def _handle_teleport_command(
self, value: str | None = None, show_message: bool = True
) -> None:
has_history = any(msg.role != Role.system for msg in self.agent_loop.messages)
if not value:
if show_message:
await self._mount_and_scroll(UserMessage("/teleport"))
if not has_history:
await self._mount_and_scroll(
ErrorMessage(
"No conversation history to teleport.",
collapsed=self._tools_collapsed,
)
)
return
elif show_message:
await self._mount_and_scroll(UserMessage(value))
self.run_worker(self._teleport(value), exclusive=False)
async def _teleport(self, prompt: str | None = None) -> None:
loading_area = self._cached_loading_area or self.query_one(
"#loading-area-content"
)
loading = LoadingWidget()
await loading_area.mount(loading)
teleport_msg = TeleportMessage()
await self._mount_and_scroll(teleport_msg)
try:
gen = self.agent_loop.teleport_to_vibe_nuage(prompt)
async for event in gen:
match event:
case TeleportCheckingGitEvent():
teleport_msg.set_status("Checking git status...")
case TeleportPushRequiredEvent(unpushed_count=count):
await loading.remove()
response = await self._ask_push_approval(count)
await loading_area.mount(loading)
teleport_msg.set_status("Teleporting...")
await gen.asend(response)
case TeleportPushingEvent():
teleport_msg.set_status("Pushing to remote...")
case TeleportAuthRequiredEvent(
user_code=code, verification_uri=uri
):
teleport_msg.set_status(
f"GitHub auth required. Code: {code} (copied)\nOpen: {uri}"
)
case TeleportAuthCompleteEvent():
teleport_msg.set_status("GitHub authenticated.")
case TeleportStartingWorkflowEvent():
teleport_msg.set_status("Starting Nuage workflow...")
case TeleportSendingGithubTokenEvent():
teleport_msg.set_status("Sending encrypted GitHub token...")
case TeleportCompleteEvent(url=url):
teleport_msg.set_complete(url)
except TeleportError as e:
await teleport_msg.remove()
await self._mount_and_scroll(
ErrorMessage(str(e), collapsed=self._tools_collapsed)
)
finally:
if loading.parent:
await loading.remove()
async def _ask_push_approval(self, count: int) -> TeleportPushResponseEvent:
word = f"commit{'s' if count != 1 else ''}"
push_label = "Push and continue"
result = await self._user_input_callback(
AskUserQuestionArgs(
questions=[
Question(
question=f"You have {count} unpushed {word}. Push to continue?",
header="Push",
options=[Choice(label=push_label), Choice(label="Cancel")],
hide_other=True,
)
]
)
)
ok = (
isinstance(result, AskUserQuestionResult)
and not result.cancelled
and bool(result.answers)
and result.answers[0].answer == push_label
)
return TeleportPushResponseEvent(approved=ok)
async def _interrupt_agent_loop(self) -> None:
if not self._agent_running or self._interrupt_requested:
@ -637,14 +785,15 @@ class VibeApp(App): # noqa: PLR0904
pass
if self.event_handler:
self.event_handler.stop_current_tool_call()
self.event_handler.stop_current_tool_call(success=False)
self.event_handler.stop_current_compact()
self._agent_running = False
loading_area = self.query_one("#loading-area-content")
loading_area = self._cached_loading_area or self.query_one(
"#loading-area-content"
)
await loading_area.remove_children()
self._loading_widget = None
self._hide_todo_area()
await self._finalize_current_streaming_message()
await self._mount_and_scroll(InterruptMessage())
@ -676,10 +825,16 @@ class VibeApp(App): # noqa: PLR0904
async def _reload_config(self) -> None:
try:
self._windowing.reset()
self._tool_call_map = None
self._history_widget_indices = WeakKeyDictionary()
await self._load_more.hide()
base_config = VibeConfig.load()
await self.agent_loop.reload_with_initial_messages(base_config=base_config)
if self._banner:
self._banner.set_state(base_config, self.agent_loop.skill_manager)
await self._mount_and_scroll(UserCommandMessage("Configuration reloaded."))
except Exception as e:
await self._mount_and_scroll(
@ -690,18 +845,19 @@ class VibeApp(App): # noqa: PLR0904
async def _clear_history(self) -> None:
try:
self._windowing.reset()
self._tool_call_map = None
self._history_widget_indices = WeakKeyDictionary()
await self.agent_loop.clear_history()
await self._finalize_current_streaming_message()
messages_area = self.query_one("#messages")
messages_area = self._cached_messages_area or self.query_one("#messages")
await messages_area.remove_children()
todo_area = self.query_one("#todo-area")
await todo_area.remove_children()
await messages_area.mount(UserMessage("/clear"))
await self._mount_and_scroll(
UserCommandMessage("Conversation history cleared!")
)
chat = self.query_one("#chat", VerticalScroll)
chat = self._cached_chat or self.query_one("#chat", ChatScroll)
chat.scroll_home(animate=False)
except Exception as e:
@ -805,16 +961,14 @@ class VibeApp(App): # noqa: PLR0904
if result.success:
if result.requires_restart:
message = f"{result.message or 'Set up Shift+Enter keybind'} (You may need to restart your terminal.)"
await self._mount_and_scroll(
UserCommandMessage(
f"{result.terminal.value}: Set up Shift+Enter keybind (You may need to restart your terminal.)"
)
UserCommandMessage(f"{result.terminal.value}: {message}")
)
else:
message = result.message or "Shift+Enter keybind already set up"
await self._mount_and_scroll(
WarningMessage(
f"{result.terminal.value}: Shift+Enter keybind already set up"
)
WarningMessage(f"{result.terminal.value}: {message}")
)
else:
await self._mount_and_scroll(
@ -828,9 +982,6 @@ class VibeApp(App): # noqa: PLR0904
self._chat_input_container.display = False
self._chat_input_container.disabled = True
if self._agent_indicator:
self._agent_indicator.display = False
self._current_bottom_app = BottomApp[type(widget).__name__.removesuffix("App")]
await bottom_container.mount(widget)
@ -843,9 +994,7 @@ class VibeApp(App): # noqa: PLR0904
return
await self._mount_and_scroll(UserCommandMessage("Configuration opened..."))
await self._switch_from_input(
ConfigApp(self.config, has_terminal_theme=self._terminal_theme is not None)
)
await self._switch_from_input(ConfigApp(self.config))
async def _switch_to_approval_app(
self, tool_name: str, tool_args: BaseModel
@ -866,9 +1015,6 @@ class VibeApp(App): # noqa: PLR0904
except Exception:
pass
if self._agent_indicator:
self._agent_indicator.display = True
if self._chat_input_container:
self._chat_input_container.disabled = False
self._chat_input_container.display = True
@ -942,12 +1088,44 @@ class VibeApp(App): # noqa: PLR0904
self._scroll_to_bottom()
self._focus_current_bottom_app()
async def on_history_load_more_requested(self, _: HistoryLoadMoreRequested) -> None:
self._load_more.set_enabled(False)
try:
if not self._windowing.has_backfill:
await self._load_more.hide()
return
if (batch := self._windowing.next_load_more_batch()) is None:
await self._load_more.hide()
return
messages_area = self._cached_messages_area or self.query_one("#messages")
if self._tool_call_map is None:
self._tool_call_map = {}
if self._load_more.widget:
before: Widget | int | None = None
after: Widget | None = self._load_more.widget
else:
before = 0
after = None
await self._mount_history_batch(
batch.messages,
messages_area,
self._tool_call_map,
start_index=batch.start_index,
before=before,
after=after,
)
if not self._windowing.has_backfill:
await self._load_more.hide()
else:
await self._load_more.show(messages_area, self._windowing.remaining)
finally:
self._load_more.set_enabled(True)
async def action_toggle_tool(self) -> None:
self._tools_collapsed = not self._tools_collapsed
for result in self.query(ToolResultMessage):
if result.tool_name != "todo":
await result.set_collapsed(self._tools_collapsed)
await result.set_collapsed(self._tools_collapsed)
try:
for error_msg in self.query(ErrorMessage):
@ -955,13 +1133,6 @@ class VibeApp(App): # noqa: PLR0904
except Exception:
pass
async def action_toggle_todo(self) -> None:
self._todos_collapsed = not self._todos_collapsed
for result in self.query(ToolResultMessage):
if result.tool_name == "todo":
await result.set_collapsed(self._todos_collapsed)
def action_cycle_mode(self) -> None:
if self._current_bottom_app != BottomApp.Input:
return
@ -973,10 +1144,9 @@ class VibeApp(App): # noqa: PLR0904
self._update_profile_widgets(self.agent_loop.agent_profile)
def _update_profile_widgets(self, profile: AgentProfile) -> None:
if self._agent_indicator:
self._agent_indicator.set_profile(profile)
if self._chat_input_container:
self._chat_input_container.set_safety(profile.safety)
self._chat_input_container.set_agent_name(profile.display_name.lower())
async def _cycle_agent(self) -> None:
new_profile = self.agent_loop.agent_manager.next_agent(
@ -1005,7 +1175,7 @@ class VibeApp(App): # noqa: PLR0904
def action_scroll_chat_up(self) -> None:
try:
chat = self.query_one("#chat", VerticalScroll)
chat = self._cached_chat or self.query_one("#chat", ChatScroll)
chat.scroll_relative(y=-5, animate=False)
self._auto_scroll = False
except Exception:
@ -1013,7 +1183,7 @@ class VibeApp(App): # noqa: PLR0904
def action_scroll_chat_down(self) -> None:
try:
chat = self.query_one("#chat", VerticalScroll)
chat = self._cached_chat or self.query_one("#chat", ChatScroll)
chat.scroll_relative(y=5, animate=False)
if self._is_scrolled_to_bottom(chat):
self._auto_scroll = True
@ -1039,42 +1209,37 @@ class VibeApp(App): # noqa: PLR0904
content = load_whats_new_content()
if content is not None:
await self._mount_and_scroll(WhatsNewMessage(content))
whats_new_message = WhatsNewMessage(content)
plan_offer = await self._plan_offer_cta()
if plan_offer is not None:
whats_new_message = WhatsNewMessage(f"{content}\n\n{plan_offer}")
if self._history_widget_indices:
whats_new_message.add_class("after-history")
await self._mount_and_scroll(whats_new_message)
await mark_version_as_seen(self._current_version, self._update_cache_repository)
async def _maybe_show_plan_offer(self) -> None:
if self._plan_offer_shown:
return
action, plan_type = await self._resolve_plan_offer_action()
self.plan_type = plan_type
if action is PlanOfferAction.NONE:
return
url = ACTION_TO_URL[action]
match action:
case PlanOfferAction.UPGRADE:
text = f"Upgrade to [Pro]({url})"
case PlanOfferAction.SWITCH_TO_PRO_KEY:
text = f"Switch to your [Pro API key]({url})"
await self._mount_and_scroll(PlanOfferMessage(text))
self._plan_offer_shown = True
async def _plan_offer_cta(self) -> str | None:
self.plan_type = PlanType.UNKNOWN
if self._plan_offer_gateway is None:
return
async def _resolve_plan_offer_action(self) -> tuple[PlanOfferAction, PlanType]:
try:
active_model = self.config.get_active_model()
provider = self.config.get_provider_for_model(active_model)
except ValueError:
return PlanOfferAction.NONE, PlanType.UNKNOWN
api_key_env = provider.api_key_env_var
api_key = getenv(api_key_env) if api_key_env else None
gateway = self._plan_offer_gateway or HttpWhoAmIGateway()
try:
return await decide_plan_offer(api_key, gateway)
api_key = resolve_api_key_for_plan(provider)
action, plan_type = await decide_plan_offer(
api_key, self._plan_offer_gateway
)
self.plan_type = plan_type
return plan_offer_cta(action)
except Exception as exc:
logger.warning(
"Plan-offer check failed (%s).", type(exc).__name__, exc_info=True
)
return PlanOfferAction.NONE, PlanType.UNKNOWN
return
async def _finalize_current_streaming_message(self) -> None:
if self._current_streaming_reasoning is not None:
@ -1108,9 +1273,10 @@ class VibeApp(App): # noqa: PLR0904
return widget
async def _mount_and_scroll(self, widget: Widget) -> None:
messages_area = self.query_one("#messages")
chat = self.query_one("#chat", VerticalScroll)
messages_area = self._cached_messages_area or self.query_one("#messages")
chat = self._cached_chat or self.query_one("#chat", ChatScroll)
was_at_bottom = self._is_scrolled_to_bottom(chat)
result: Widget | None = None
if was_at_bottom:
self._auto_scroll = True
@ -1140,12 +1306,15 @@ class VibeApp(App): # noqa: PLR0904
else:
await self._finalize_current_streaming_message()
await messages_area.mount(widget)
result = widget
is_tool_message = isinstance(widget, (ToolCallMessage, ToolResultMessage))
if not is_tool_message:
self.call_after_refresh(self._scroll_to_bottom)
if result is not None:
self.call_after_refresh(self._try_prune)
if was_at_bottom:
self.call_after_refresh(self._anchor_if_scrollable)
@ -1158,7 +1327,7 @@ class VibeApp(App): # noqa: PLR0904
def _scroll_to_bottom(self) -> None:
try:
chat = self.query_one("#chat")
chat = self._cached_chat or self.query_one("#chat", ChatScroll)
chat.scroll_end(animate=False)
except Exception:
pass
@ -1166,17 +1335,50 @@ class VibeApp(App): # noqa: PLR0904
def _scroll_to_bottom_deferred(self) -> None:
self.call_after_refresh(self._scroll_to_bottom)
async def _try_prune(self) -> None:
messages_area = self._cached_messages_area or self.query_one("#messages")
await prune_by_height(messages_area, PRUNE_LOW_MARK, PRUNE_HIGH_MARK)
if self._load_more.widget and not self._load_more.widget.parent:
self._load_more.widget = None
async def _refresh_windowing_from_history(self) -> None:
if self._load_more.widget is None:
return
messages_area = self._cached_messages_area or self.query_one("#messages")
has_backfill, tool_call_map = sync_backfill_state(
history_messages=non_system_history_messages(self.agent_loop.messages),
messages_children=list(messages_area.children),
history_widget_indices=self._history_widget_indices,
windowing=self._windowing,
)
self._tool_call_map = tool_call_map
await self._load_more.set_visible(
messages_area, visible=has_backfill, remaining=self._windowing.remaining
)
def _anchor_if_scrollable(self) -> None:
if not self._auto_scroll:
return
try:
chat = self.query_one("#chat", VerticalScroll)
chat = self._cached_chat or self.query_one("#chat", ChatScroll)
if chat.max_scroll_y == 0:
return
chat.anchor()
except Exception:
pass
def _align_chat_after_history_rebuild(self, has_backfill: bool) -> None:
try:
chat = self._cached_chat or self.query_one("#chat", ChatScroll)
if has_backfill and chat.max_scroll_y > 0:
chat.anchor(True)
chat.scroll_end(animate=False)
chat.anchor(False)
return
chat.scroll_end(animate=False)
except Exception:
pass
def _schedule_update_notification(self) -> None:
if self._update_notifier is None or not self.config.enable_update_checks:
return
@ -1227,8 +1429,12 @@ class VibeApp(App): # noqa: PLR0904
message, title="Update available", severity="information", timeout=10
)
def action_copy_selection(self) -> None:
copy_selection_to_clipboard(self, show_toast=False)
def on_mouse_up(self, event: MouseUp) -> None:
copy_selection_to_clipboard(self)
if self.config.autocopy_to_clipboard:
copy_selection_to_clipboard(self, show_toast=True)
def on_app_blur(self, event: AppBlur) -> None:
if self._chat_input_container and self._chat_input_container.input_widget:
@ -1248,14 +1454,21 @@ def _print_session_resume_message(session_id: str | None) -> None:
print(f"Or: vibe --resume {session_id}")
def run_textual_ui(agent_loop: AgentLoop, initial_prompt: str | None = None) -> None:
def run_textual_ui(
agent_loop: AgentLoop,
initial_prompt: str | None = None,
teleport_on_start: bool = False,
) -> None:
update_notifier = PyPIUpdateGateway(project_name="mistral-vibe")
update_cache_repository = FileSystemUpdateCacheRepository()
plan_offer_gateway = HttpWhoAmIGateway()
app = VibeApp(
agent_loop=agent_loop,
initial_prompt=initial_prompt,
teleport_on_start=teleport_on_start,
update_notifier=update_notifier,
update_cache_repository=update_cache_repository,
plan_offer_gateway=plan_offer_gateway,
)
session_id = app.run()
_print_session_resume_message(session_id)

823
vibe/cli/textual_ui/app.tcss
View file

File diff suppressed because it is too large Load diff

37
vibe/cli/textual_ui/handlers/event_handler.py
View file

@ -7,6 +7,7 @@ from vibe.cli.textual_ui.widgets.compact import CompactMessage
from vibe.cli.textual_ui.widgets.messages import AssistantMessage, ReasoningMessage
from vibe.cli.textual_ui.widgets.no_markup_static import NoMarkupStatic
from vibe.cli.textual_ui.widgets.tools import ToolCallMessage, ToolResultMessage
from vibe.core.tools.ui import ToolUIDataAdapter
from vibe.core.types import (
AssistantEvent,
BaseEvent,
@ -29,15 +30,11 @@ class EventHandler:
self,
mount_callback: Callable,
scroll_callback: Callable,
todo_area_callback: Callable,
get_tools_collapsed: Callable[[], bool],
get_todos_collapsed: Callable[[], bool],
) -> None:
self.mount_callback = mount_callback
self.scroll_callback = scroll_callback
self.todo_area_callback = todo_area_callback
self.get_tools_collapsed = get_tools_collapsed
self.get_todos_collapsed = get_todos_collapsed
self.current_tool_call: ToolCallMessage | None = None
self.current_compact: CompactMessage | None = None
@ -93,35 +90,21 @@ class EventHandler:
tool_call = ToolCallMessage(event)
if loading_widget and event.tool_class:
from vibe.core.tools.ui import ToolUIDataAdapter
adapter = ToolUIDataAdapter(event.tool_class)
status_text = adapter.get_status_text()
loading_widget.set_status(status_text)
# Don't show todo in messages
if event.tool_name != "todo":
await self.mount_callback(tool_call)
self.current_tool_call = tool_call
await self.mount_callback(tool_call)
return tool_call
async def _handle_tool_result(self, event: ToolResultEvent) -> None:
if event.tool_name == "todo":
todos_collapsed = self.get_todos_collapsed()
tool_result = ToolResultMessage(
event, self.current_tool_call, collapsed=todos_collapsed
)
# Show in todo area
todo_area = self.todo_area_callback()
await todo_area.remove_children()
await todo_area.mount(tool_result)
else:
tools_collapsed = self.get_tools_collapsed()
tool_result = ToolResultMessage(
event, self.current_tool_call, collapsed=tools_collapsed
)
await self.mount_callback(tool_result)
tools_collapsed = self.get_tools_collapsed()
tool_result = ToolResultMessage(
event, self.current_tool_call, collapsed=tools_collapsed
)
await self.mount_callback(tool_result)
self.current_tool_call = None
@ -153,9 +136,9 @@ class EventHandler:
async def _handle_unknown_event(self, event: BaseEvent) -> None:
await self.mount_callback(NoMarkupStatic(str(event), classes="unknown-event"))
def stop_current_tool_call(self) -> None:
def stop_current_tool_call(self, success: bool = True) -> None:
if self.current_tool_call:
self.current_tool_call.stop_spinning()
self.current_tool_call.stop_spinning(success=success)
self.current_tool_call = None
def stop_current_compact(self) -> None:

266
vibe/cli/textual_ui/terminal_theme.py
View file

@ -1,266 +0,0 @@
from __future__ import annotations
from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass, fields
import os
import re
import sys
from typing import Any
from textual.theme import Theme
try:
import select
import termios
_UNIX_AVAILABLE = True
except ImportError:
select = None # type: ignore[assignment]
termios: Any = None
_UNIX_AVAILABLE = False
TERMINAL_THEME_NAME = "terminal"
_LUMINANCE_THRESHOLD = 0.5
_RGB_16BIT_LEN = 4
_RGB_8BIT_LEN = 2
# OSC codes for terminal colors
_OSC_FOREGROUND = "10"
_OSC_BACKGROUND = "11"
# ANSI color indices (0-15) mapped to field names
_ANSI_COLORS = (
"black",
"red",
"green",
"yellow",
"blue",
"magenta",
"cyan",
"white",
"bright_black",
"bright_red",
"bright_green",
"bright_yellow",
"bright_blue",
"bright_magenta",
"bright_cyan",
"bright_white",
)
# DA1 (Primary Device Attributes) query - used to detect unsupported terminals
# Terminals respond to DA1 even if they don't support OSC color queries
_DA1_QUERY = b"\x1b[c"
_DA1_RESPONSE_PREFIX = b"\x1b[?"
_OSC_RESPONSE_RE = re.compile(
rb"\x1b\](10|11|4;\d+);rgb:([0-9a-fA-F]+)/([0-9a-fA-F]+)/([0-9a-fA-F]+)(?:\x1b\\|\x07)"
)
@dataclass
class TerminalColors:
foreground: str | None = None
background: str | None = None
black: str | None = None
red: str | None = None
green: str | None = None
yellow: str | None = None
blue: str | None = None
magenta: str | None = None
cyan: str | None = None
white: str | None = None
bright_black: str | None = None
bright_red: str | None = None
bright_green: str | None = None
bright_yellow: str | None = None
bright_blue: str | None = None
bright_magenta: str | None = None
bright_cyan: str | None = None
bright_white: str | None = None
def is_complete(self) -> bool:
return all(getattr(self, f.name) is not None for f in fields(self))
def _build_osc_query(code: str) -> bytes:
"""Build an OSC query: ESC ] <code> ; ? ST"""
return f"\x1b]{code};?\x1b\\".encode()
def _build_color_queries() -> tuple[bytes, dict[bytes, str]]:
"""Build all OSC color queries and the mapping from OSC codes to field names."""
queries = bytearray()
osc_to_field: dict[bytes, str] = {}
# Foreground and background
queries.extend(_build_osc_query(_OSC_FOREGROUND))
queries.extend(_build_osc_query(_OSC_BACKGROUND))
osc_to_field[_OSC_FOREGROUND.encode()] = "foreground"
osc_to_field[_OSC_BACKGROUND.encode()] = "background"
# ANSI colors 0-15
for i, name in enumerate(_ANSI_COLORS):
code = f"4;{i}"
queries.extend(_build_osc_query(code))
osc_to_field[code.encode()] = name
return bytes(queries), osc_to_field
_COLOR_QUERIES, _OSC_TO_FIELD = _build_color_queries()
@contextmanager
def _raw_mode(fd: int) -> Iterator[None]:
"""Context manager to temporarily set terminal to raw mode."""
assert termios is not None # Only called on Unix so typing doesn't freak out
try:
old_settings = termios.tcgetattr(fd)
except termios.error:
yield
return
try:
new_settings = termios.tcgetattr(fd)
new_settings[3] &= ~(termios.ECHO | termios.ICANON)
termios.tcsetattr(fd, termios.TCSADRAIN, new_settings)
yield
finally:
try:
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
except termios.error:
pass
def _parse_rgb(r_hex: bytes, g_hex: bytes, b_hex: bytes) -> str | None:
"""Parse RGB hex values to a #rrggbb string.
Terminals return either 16-bit (4 hex chars) or 8-bit (2 hex chars) per channel.
"""
try:
if len(r_hex) == _RGB_16BIT_LEN:
r, g, b = int(r_hex[:2], 16), int(g_hex[:2], 16), int(b_hex[:2], 16)
elif len(r_hex) == _RGB_8BIT_LEN:
r, g, b = int(r_hex, 16), int(g_hex, 16), int(b_hex, 16)
else:
return None
return f"#{r:02x}{g:02x}{b:02x}"
except ValueError:
return None
def _read_responses(fd: int, timeout: float = 1.0) -> bytes:
"""Read terminal responses until DA1 response or timeout.
Uses the DA1 trick: we send color queries followed by DA1. Since terminals
respond in order, receiving the DA1 response means all color responses
(if any) have been received.
"""
assert select is not None # Only called on Unix so typing doesn't freak out
response = bytearray()
while True:
ready, _, _ = select.select([fd], [], [], timeout)
if not ready:
break
chunk = os.read(fd, 4096)
if not chunk:
break
response.extend(chunk)
# DA1 response received - we have all the color responses
if _DA1_RESPONSE_PREFIX in response:
break
return bytes(response)
def _parse_osc_responses(response: bytes) -> TerminalColors:
colors = TerminalColors()
for match in _OSC_RESPONSE_RE.finditer(response):
osc_code, r_hex, g_hex, b_hex = match.groups()
field = _OSC_TO_FIELD.get(osc_code)
if field and (color := _parse_rgb(r_hex, g_hex, b_hex)):
setattr(colors, field, color)
return colors
def _query_terminal_colors() -> TerminalColors:
if not _UNIX_AVAILABLE:
return TerminalColors()
if not sys.stdin.isatty() or not sys.stdout.isatty():
return TerminalColors()
fd = sys.stdin.fileno()
try:
with _raw_mode(fd):
os.write(sys.stdout.fileno(), _COLOR_QUERIES + _DA1_QUERY)
response = _read_responses(fd)
return _parse_osc_responses(response)
except OSError:
return TerminalColors()
def _hex_to_rgb(hex_color: str) -> tuple[int, int, int]:
h = hex_color.lstrip("#")
return int(h[0:2], 16), int(h[2:4], 16), int(h[4:6], 16)
def _rgb_to_hex(r: int, g: int, b: int) -> str:
return f"#{r:02x}{g:02x}{b:02x}"
def _adjust_brightness(hex_color: str, factor: float) -> str:
r, g, b = _hex_to_rgb(hex_color)
return _rgb_to_hex(
min(255, max(0, int(r * factor))),
min(255, max(0, int(g * factor))),
min(255, max(0, int(b * factor))),
)
def _blend(c1: str, c2: str, ratio: float = 0.5) -> str:
r1, g1, b1 = _hex_to_rgb(c1)
r2, g2, b2 = _hex_to_rgb(c2)
return _rgb_to_hex(
int(r1 * (1 - ratio) + r2 * ratio),
int(g1 * (1 - ratio) + g2 * ratio),
int(b1 * (1 - ratio) + b2 * ratio),
)
def _luminance(hex_color: str) -> float:
"""Calculate perceived luminance (0-1) using ITU-R BT.601 coefficients."""
r, g, b = _hex_to_rgb(hex_color)
return (0.299 * r + 0.587 * g + 0.114 * b) / 255
def capture_terminal_theme() -> Theme | None:
colors = _query_terminal_colors()
if not colors.background or not colors.foreground:
return None
is_dark = _luminance(colors.background) < _LUMINANCE_THRESHOLD
fg = colors.foreground
bg = colors.background
surface = _adjust_brightness(bg, 1.15 if is_dark else 0.95)
panel = _blend(bg, surface)
return Theme(
name=TERMINAL_THEME_NAME,
primary=colors.blue or fg,
secondary=colors.cyan or fg,
warning=colors.yellow or fg,
error=colors.red or fg,
success=colors.green or fg,
accent=colors.magenta or fg,
foreground=fg,
background=bg,
surface=surface,
panel=panel,
dark=is_dark,
)

41
vibe/cli/textual_ui/widgets/agent_indicator.py
View file

@ -1,41 +0,0 @@
from __future__ import annotations
from textual.widgets import Static
from vibe.core.agents import AgentProfile, AgentSafety, BuiltinAgentName
AGENT_ICONS: dict[str, str] = {
BuiltinAgentName.DEFAULT: "⏵",
BuiltinAgentName.PLAN: "⏸",
BuiltinAgentName.ACCEPT_EDITS: "⏵⏵",
BuiltinAgentName.AUTO_APPROVE: "⏵⏵⏵",
}
SAFETY_CLASSES: dict[AgentSafety, str] = {
AgentSafety.SAFE: "agent-safe",
AgentSafety.NEUTRAL: "agent-neutral",
AgentSafety.DESTRUCTIVE: "agent-destructive",
AgentSafety.YOLO: "agent-yolo",
}
class AgentIndicator(Static):
def __init__(self, profile: AgentProfile) -> None:
super().__init__(markup=False)
self.can_focus = False
self.profile = profile
self._update_display()
def _update_display(self) -> None:
icon = AGENT_ICONS.get(self.profile.name, "")
name = self.profile.display_name.lower()
self.update(f"{icon}{' ' if icon else ''}{name} agent (shift+tab to cycle)")
for safety_class in SAFETY_CLASSES.values():
self.remove_class(safety_class)
self.add_class(SAFETY_CLASSES[self.profile.safety])
def set_profile(self, profile: AgentProfile) -> None:
self.profile = profile
self._update_display()

26
vibe/cli/textual_ui/widgets/approval_app.py
View file

@ -66,6 +66,18 @@ class ApprovalApp(Container):
self.help_widget: Static | None = None
def compose(self) -> ComposeResult:
with Vertical(id="approval-options"):
yield NoMarkupStatic("")
for _ in range(3):
widget = NoMarkupStatic("", classes="approval-option")
self.option_widgets.append(widget)
yield widget
yield NoMarkupStatic("")
self.help_widget = NoMarkupStatic(
"↑↓ navigate Enter select ESC reject", classes="approval-help"
)
yield self.help_widget
with Vertical(id="approval-content"):
self.title_widget = NoMarkupStatic(
f"⚠ {self.tool_name} command", classes="approval-title"
@ -78,20 +90,6 @@ class ApprovalApp(Container):
)
yield self.tool_info_container
yield NoMarkupStatic("")
for _ in range(3):
widget = NoMarkupStatic("", classes="approval-option")
self.option_widgets.append(widget)
yield widget
yield NoMarkupStatic("")
self.help_widget = NoMarkupStatic(
"↑↓ navigate Enter select ESC reject", classes="approval-help"
)
yield self.help_widget
async def on_mount(self) -> None:
await self._update_tool_info()
self._update_options()

85
vibe/cli/textual_ui/widgets/banner/banner.py Normal file
View file

@ -0,0 +1,85 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from textual.app import ComposeResult
from textual.containers import Horizontal, Vertical
from textual.reactive import reactive
from textual.widgets import Static
from vibe import __version__
from vibe.cli.textual_ui.widgets.banner.petit_chat import PetitChat
from vibe.cli.textual_ui.widgets.no_markup_static import NoMarkupStatic
from vibe.core.config import VibeConfig
from vibe.core.skills.manager import SkillManager
@dataclass
class BannerState:
active_model: str = ""
models_count: int = 0
mcp_servers_count: int = 0
skills_count: int = 0
class Banner(Static):
state = reactive(BannerState(), init=False)
def __init__(
self, config: VibeConfig, skill_manager: SkillManager, **kwargs: Any
) -> None:
super().__init__(**kwargs)
self.can_focus = False
self._initial_state = BannerState(
active_model=config.active_model,
models_count=len(config.models),
mcp_servers_count=len(config.mcp_servers),
skills_count=len(skill_manager.available_skills),
)
self._animated = not config.disable_welcome_banner_animation
def compose(self) -> ComposeResult:
with Horizontal(id="banner-container"):
yield PetitChat(animate=self._animated)
with Vertical(id="banner-info"):
with Horizontal(classes="banner-line"):
yield NoMarkupStatic("Mistral Vibe", id="banner-brand")
yield NoMarkupStatic(" ", classes="banner-spacer")
yield NoMarkupStatic(f"v{__version__} · ", classes="banner-meta")
yield NoMarkupStatic("", id="banner-model")
with Horizontal(classes="banner-line"):
yield NoMarkupStatic("", id="banner-meta-counts")
with Horizontal(classes="banner-line"):
yield NoMarkupStatic("Type ", classes="banner-meta")
yield NoMarkupStatic("/help", classes="banner-cmd")
yield NoMarkupStatic(" for more information", classes="banner-meta")
def on_mount(self) -> None:
self.state = self._initial_state
def watch_state(self) -> None:
self.query_one("#banner-model", NoMarkupStatic).update(self.state.active_model)
self.query_one("#banner-meta-counts", NoMarkupStatic).update(
self._format_meta_counts()
)
def freeze_animation(self) -> None:
if self._animated:
self.query_one(PetitChat).freeze_animation()
def set_state(self, config: VibeConfig, skill_manager: SkillManager) -> None:
self.state = BannerState(
active_model=config.active_model,
models_count=len(config.models),
mcp_servers_count=len(config.mcp_servers),
skills_count=len(skill_manager.available_skills),
)
def _format_meta_counts(self) -> str:
return (
f"{self.state.models_count} model{'s' if self.state.models_count != 1 else ''}"
f" · {self.state.mcp_servers_count} MCP server{'s' if self.state.mcp_servers_count != 1 else ''}"
f" · {self.state.skills_count} skill{'s' if self.state.skills_count != 1 else ''}"
)

195
vibe/cli/textual_ui/widgets/banner/petit_chat.py Normal file
View file

@ -0,0 +1,195 @@
from __future__ import annotations
from typing import Any
from textual.app import ComposeResult
from textual.timer import Timer
from textual.widgets import Static
from vibe.cli.textual_ui.widgets.braille_renderer import render_braille
WIDTH = 22
HEIGHT = 12
STARTING_DOTS = [
set[int](),
{6, 7, 15, 19},
{5, 8, 14, 16, 18, 20},
{4, 6, 7, 14, 17, 20},
{3, 5, 10, 11, 12, 14, 20},
{3, 5, 9, 13, 14, 16, 18, 20},
{3, 5, 8, 13, 17, 21},
{3, 6, 7, 8, 11, 14, 15, 16, 18, 19, 20},
{4, 5, 8, 12, 17, 19},
{6, 7, 8, 13, 18, 20},
{9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20},
set[int](),
]
QUEUE_RIGHT_TO_MID = {
"remove": {1j + 6, 1j + 7, 2j + 8, 3j + 4, 3j + 6, 3j + 7, 8j + 4, 8j + 5},
"add": {1j + 4, 2j + 3, 3j + 3, 3j + 5, 7j + 5, 8j + 3, 9j + 4, 9j + 5},
}
QUEUE_MID_TO_RIGHT = {
"remove": QUEUE_RIGHT_TO_MID["add"],
"add": QUEUE_RIGHT_TO_MID["remove"],
}
QUEUE_MID_TO_LEFT = {
"remove": {1j + 4, 2j + 5, 3j + 3, 3j + 5, 7j + 5, 8j + 3, 9j + 4, 9j + 5},
"add": {1j + 1, 1j + 2, 2j, 3j + 1, 3j + 2, 3j + 4, 8j + 4, 8j + 5},
}
QUEUE_LEFT_TO_MID = {
"remove": QUEUE_MID_TO_LEFT["add"],
"add": QUEUE_MID_TO_LEFT["remove"],
}
WAIT = {"remove": set[int](), "add": set[int]()}
HEAD_RIGHT = {"remove": {5j + 16, 5j + 18, 6j + 17}, "add": {5j + 17, 5j + 19, 6j + 18}}
HEAD_LEFT = {"remove": {5j + 17, 5j + 19, 6j + 18}, "add": {5j + 16, 5j + 18, 6j + 17}}
HEAD_DOWN = {
"remove": {
1j + 15,
1j + 19,
2j + 14,
2j + 16,
2j + 18,
2j + 20,
3j + 17,
5j + 17,
5j + 19,
6j + 13,
6j + 18,
6j + 21,
7j + 14,
7j + 15,
7j + 16,
7j + 19,
7j + 20,
},
"add": {
2j + 15,
2j + 19,
3j + 16,
3j + 18,
4j + 17,
6j + 14,
6j + 17,
6j + 19,
6j + 20,
7j + 13,
7j + 18,
7j + 21,
8j + 14,
8j + 15,
8j + 16,
8j + 18,
8j + 20,
},
}
HEAD_UP = {
"remove": {
2j + 15,
2j + 19,
3j + 16,
3j + 18,
4j + 17,
6j + 14,
6j + 17,
6j + 19,
6j + 20,
7j + 13,
7j + 18,
7j + 21,
8j + 14,
8j + 15,
8j + 16,
8j + 18,
8j + 20,
},
"add": {
1j + 15,
1j + 19,
2j + 14,
2j + 16,
2j + 18,
2j + 20,
3j + 17,
5j + 17,
5j + 19,
6j + 13,
6j + 18,
6j + 21,
7j + 14,
7j + 15,
7j + 16,
7j + 18,
7j + 19,
7j + 20,
},
}
BLINK_EYES_HEAD_HIGH = [
{"remove": {5j + 16, 5j + 18}, "add": set[int]()},
{"remove": set[int](), "add": {5j + 16, 5j + 18}},
]
BLINK_EYES_HEAD_LOW = [
{"remove": {6j + 17, 6j + 19}, "add": set[int]()},
{"remove": set[int](), "add": {6j + 17, 6j + 19}},
]
TRANSITIONS = [
*BLINK_EYES_HEAD_HIGH,
WAIT,
QUEUE_RIGHT_TO_MID,
HEAD_RIGHT,
WAIT,
QUEUE_MID_TO_LEFT,
WAIT,
QUEUE_LEFT_TO_MID,
WAIT,
HEAD_DOWN,
WAIT,
QUEUE_MID_TO_RIGHT,
*BLINK_EYES_HEAD_LOW,
WAIT,
QUEUE_RIGHT_TO_MID,
WAIT,
QUEUE_MID_TO_LEFT,
WAIT,
HEAD_UP,
WAIT,
QUEUE_LEFT_TO_MID,
HEAD_LEFT,
WAIT,
QUEUE_MID_TO_RIGHT,
]
# cf render_braille() docstring for coordinates convention
class PetitChat(Static):
def __init__(self, animate: bool = True, **kwargs: Any) -> None:
super().__init__(**kwargs, classes="banner-chat")
self._dots = {1j * y + x for y, row in enumerate(STARTING_DOTS) for x in row}
self._transition_index = 0
self._do_animate = animate
self._freeze_requested = False
self._timer: Timer | None = None
def compose(self) -> ComposeResult:
yield Static(render_braille(self._dots, WIDTH, HEIGHT), classes="petit-chat")
def on_mount(self) -> None:
self._inner = self.query_one(".petit-chat", Static)
if self._do_animate:
self._timer = self.set_interval(0.16, self._apply_next_transition)
def freeze_animation(self) -> None:
self._freeze_requested = True
def _apply_next_transition(self) -> None:
if self._freeze_requested and self._transition_index == 0:
if self._timer:
self._timer.stop()
self._timer = None
return
transition = TRANSITIONS[self._transition_index]
self._dots -= transition["remove"]
self._dots |= transition["add"]
self._transition_index = (self._transition_index + 1) % len(TRANSITIONS)
self._inner.update(render_braille(self._dots, WIDTH, HEIGHT))

58
vibe/cli/textual_ui/widgets/braille_renderer.py Normal file
View file

@ -0,0 +1,58 @@
from __future__ import annotations
from collections.abc import Iterable
import math
# for more details on braille characters encoding, see: https://en.wikipedia.org/wiki/Braille_Patterns
_BRAILLE_DOT_COUNT = 8
def _braille_dot_index(x: int, y: int) -> int:
"""returns the number associated with a dot in a braille character
x ∈ {0, 1}, y ∈ {0, 1, 2, 3}
-x->
| 1 4
y 2 5
| 3 6
V 7 8
"""
if y < 3: # noqa: PLR2004
return y + 1 + 3 * x
return 7 + x
def _braille_char_from_dot_indices(indices: list[int]) -> str:
if any(n < 1 or n > _BRAILLE_DOT_COUNT for n in indices):
raise ValueError(f"Invalid braille dot indices: {indices}")
return chr(0x2800 + sum(2 ** (d - 1) for d in indices)) if indices else " "
def render_braille(dot_coords: Iterable[complex], width: int, height: int) -> str:
"""this function receives a list of dot coordinantes, a width and a height,
and returns a string representing these dots with braille characters.
Origin is (0,0) and is located at the top left:
0----x---->
|
y
|
V
"""
dots_matrix: list[list[list[int]]] = [
[[] for _ in range(math.ceil(width / 2))] for _ in range(math.ceil(height / 4))
] # the list of dots for each character in the final str
for coord in dot_coords:
x = int(coord.real // 2)
y = int(coord.imag // 4)
sub_x = int(coord.real) % 2
sub_y = int(coord.imag) % 4
dots_matrix[y][x].append(_braille_dot_index(sub_x, sub_y))
braille_chars = [
[_braille_char_from_dot_indices(char_dots) for char_dots in row]
for row in dots_matrix
]
return "\n".join("".join(row) for row in braille_chars)

14
vibe/cli/textual_ui/widgets/chat_input/body.py
View file

@ -20,10 +20,16 @@ class ChatInputBody(Widget):
self.value = value
super().__init__()
def __init__(self, history_file: Path | None = None, **kwargs: Any) -> None:
def __init__(
self,
history_file: Path | None = None,
nuage_enabled: bool = False,
**kwargs: Any,
) -> None:
super().__init__(**kwargs)
self.input_widget: ChatTextArea | None = None
self.prompt_widget: NoMarkupStatic | None = None
self._nuage_enabled = nuage_enabled
if history_file:
self.history = HistoryManager(history_file)
@ -37,7 +43,9 @@ class ChatInputBody(Widget):
self.prompt_widget = NoMarkupStatic(">", id="prompt")
yield self.prompt_widget
self.input_widget = ChatTextArea(placeholder="Ask anything...", id="input")
self.input_widget = ChatTextArea(
id="input", nuage_enabled=self._nuage_enabled
)
yield self.input_widget
def on_mount(self) -> None:
@ -49,6 +57,8 @@ class ChatInputBody(Widget):
return "!", text[1:]
elif text.startswith("/"):
return "/", text[1:]
elif text.startswith("&") and self._nuage_enabled:
return "&", text[1:]
else:
return ">", text

22
vibe/cli/textual_ui/widgets/chat_input/container.py
View file

@ -40,14 +40,18 @@ class ChatInputContainer(Vertical):
history_file: Path | None = None,
command_registry: CommandRegistry | None = None,
safety: AgentSafety = AgentSafety.NEUTRAL,
agent_name: str = "",
skill_entries_getter: Callable[[], list[tuple[str, str]]] | None = None,
nuage_enabled: bool = False,
**kwargs: Any,
) -> None:
super().__init__(**kwargs)
self._history_file = history_file
self._command_registry = command_registry or CommandRegistry()
self._safety = safety
self._agent_name = agent_name
self._skill_entries_getter = skill_entries_getter
self._nuage_enabled = nuage_enabled
self._completion_manager = MultiCompletionManager([
SlashCommandController(CommandCompleter(self._get_slash_entries), self),
@ -71,8 +75,13 @@ class ChatInputContainer(Vertical):
yield self._completion_popup
border_class = SAFETY_BORDER_CLASSES.get(self._safety, "")
with Vertical(id=self.ID_INPUT_BOX, classes=border_class):
self._body = ChatInputBody(history_file=self._history_file, id="input-body")
with Vertical(id=self.ID_INPUT_BOX, classes=border_class) as input_box:
input_box.border_title = self._agent_name
self._body = ChatInputBody(
history_file=self._history_file,
id="input-body",
nuage_enabled=self._nuage_enabled,
)
yield self._body
@ -175,3 +184,12 @@ class ChatInputContainer(Vertical):
if safety in SAFETY_BORDER_CLASSES:
input_box.add_class(SAFETY_BORDER_CLASSES[safety])
def set_agent_name(self, name: str) -> None:
self._agent_name = name
try:
input_box = self.get_widget_by_id(self.ID_INPUT_BOX)
input_box.border_title = name
except Exception:
pass

19
vibe/cli/textual_ui/widgets/chat_input/text_area.py
View file

@ -13,7 +13,7 @@ from vibe.cli.textual_ui.widgets.chat_input.completion_manager import (
MultiCompletionManager,
)
InputMode = Literal["!", "/", ">"]
InputMode = Literal["!", "/", ">", "&"]
class ChatTextArea(TextArea):
@ -28,7 +28,6 @@ class ChatTextArea(TextArea):
Binding("ctrl+g", "open_external_editor", "External Editor", show=False),
]
MODE_CHARACTERS: ClassVar[set[Literal["!", "/"]]] = {"!", "/"}
DEFAULT_MODE: ClassVar[Literal[">"]] = ">"
class Submitted(Message):
@ -50,14 +49,15 @@ class ChatTextArea(TextArea):
"""Message sent when history navigation should be reset."""
class ModeChanged(Message):
"""Message sent when the input mode changes (>, !, /)."""
"""Message sent when the input mode changes (>, !, /, &)."""
def __init__(self, mode: InputMode) -> None:
self.mode = mode
super().__init__()
def __init__(self, **kwargs: Any) -> None:
def __init__(self, nuage_enabled: bool = False, **kwargs: Any) -> None:
super().__init__(**kwargs)
self._nuage_enabled = nuage_enabled
self._input_mode: InputMode = self.DEFAULT_MODE
self._history_prefix: str | None = None
self._last_text = ""
@ -213,7 +213,7 @@ class ChatTextArea(TextArea):
if (
event.character
and event.character in self.MODE_CHARACTERS
and event.character in self.mode_characters
and not self.text
and self._input_mode == self.DEFAULT_MODE
):
@ -328,7 +328,14 @@ class ChatTextArea(TextArea):
return self.get_cursor_offset() + self._get_mode_prefix_length()
def _get_mode_prefix_length(self) -> int:
return {">": 0, "/": 1, "!": 1}[self._input_mode]
return {">": 0, "/": 1, "!": 1, "&": 1}[self._input_mode]
@property
def mode_characters(self) -> set[InputMode]:
chars: set[InputMode] = {"!", "/"}
if self._nuage_enabled:
chars.add("&")
return chars
@property
def input_mode(self) -> InputMode:

48
vibe/cli/textual_ui/widgets/config_app.py
View file

@ -7,19 +7,13 @@ from textual.app import ComposeResult
from textual.binding import Binding, BindingType
from textual.containers import Container, Vertical
from textual.message import Message
from textual.theme import BUILTIN_THEMES
from textual.widgets import Static
from vibe.cli.textual_ui.terminal_theme import TERMINAL_THEME_NAME
from vibe.cli.textual_ui.widgets.no_markup_static import NoMarkupStatic
if TYPE_CHECKING:
from vibe.core.config import VibeConfig
_ALL_THEMES = [TERMINAL_THEME_NAME] + sorted(
k for k in BUILTIN_THEMES if k != "textual-ansi"
)
class SettingDefinition(TypedDict):
key: str
@ -46,22 +40,16 @@ class ConfigApp(Container):
self.value = value
class ConfigClosed(Message):
def __init__(self, changes: dict[str, str]) -> None:
def __init__(self, changes: dict[str, str | bool]) -> None:
super().__init__()
self.changes = changes
def __init__(self, config: VibeConfig, *, has_terminal_theme: bool = False) -> None:
def __init__(self, config: VibeConfig) -> None:
super().__init__(id="config-app")
self.config = config
self.selected_index = 0
self.changes: dict[str, str] = {}
themes = (
_ALL_THEMES
if has_terminal_theme
else [t for t in _ALL_THEMES if t != TERMINAL_THEME_NAME]
)
self.settings: list[SettingDefinition] = [
{
"key": "active_model",
@ -70,10 +58,10 @@ class ConfigApp(Container):
"options": [m.alias for m in self.config.models],
},
{
"key": "textual_theme",
"label": "Theme",
"key": "autocopy_to_clipboard",
"label": "Auto-copy",
"type": "cycle",
"options": themes,
"options": ["On", "Off"],
},
]
@ -104,6 +92,15 @@ class ConfigApp(Container):
self._update_display()
self.focus()
def _get_display_value(self, setting: SettingDefinition) -> str:
key = setting["key"]
if key in self.changes:
return self.changes[key]
raw_value = getattr(self.config, key, "")
if isinstance(raw_value, bool):
return "On" if raw_value else "Off"
return str(raw_value)
def _update_display(self) -> None:
for i, (setting, widget) in enumerate(
zip(self.settings, self.setting_widgets, strict=True)
@ -112,9 +109,7 @@ class ConfigApp(Container):
cursor = "› " if is_selected else " "
label: str = setting["label"]
value: str = self.changes.get(
setting["key"], getattr(self.config, setting["key"], "")
)
value: str = self._get_display_value(setting)
text = f"{cursor}{label}: {value}"
@ -140,7 +135,7 @@ class ConfigApp(Container):
def action_toggle_setting(self) -> None:
setting = self.settings[self.selected_index]
key: str = setting["key"]
current: str = self.changes.get(key, getattr(self.config, key)) or ""
current: str = self._get_display_value(setting)
options: list[str] = setting["options"]
new_value = ""
@ -160,8 +155,17 @@ class ConfigApp(Container):
def action_cycle(self) -> None:
self.action_toggle_setting()
def _convert_changes_for_save(self) -> dict[str, str | bool]:
result: dict[str, str | bool] = {}
for key, value in self.changes.items():
if value in {"On", "Off"}:
result[key] = value == "On"
else:
result[key] = value
return result
def action_close(self) -> None:
self.post_message(self.ConfigClosed(changes=self.changes.copy()))
self.post_message(self.ConfigClosed(changes=self._convert_changes_for_save()))
def on_blur(self, event: events.Blur) -> None:
self.call_after_refresh(self.focus)

43
vibe/cli/textual_ui/widgets/load_more.py Normal file
View file

@ -0,0 +1,43 @@
from __future__ import annotations
from textual.app import ComposeResult
from textual.containers import Horizontal
from textual.message import Message
from textual.widgets import Button, Static
class HistoryLoadMoreRequested(Message):
pass
class HistoryLoadMoreMessage(Static):
def __init__(self) -> None:
super().__init__()
self.add_class("history-load-more-message")
self._label_widget: Button | None = None
self._remaining: int | None = None
def compose(self) -> ComposeResult:
with Horizontal(classes="history-load-more-container"):
self._label_widget = Button(
self._label_text(), classes="history-load-more-button"
)
yield self._label_widget
def _label_text(self) -> str:
if self._remaining is None:
return "Load more messages"
return f"Load more messages ({self._remaining})"
def set_enabled(self, enabled: bool) -> None:
if self._label_widget:
self._label_widget.disabled = not enabled
def set_remaining(self, remaining: int | None) -> None:
self._remaining = remaining
if self._label_widget:
self._label_widget.label = self._label_text()
def on_button_pressed(self, event: Button.Pressed) -> None:
event.stop()
self.post_message(HistoryLoadMoreRequested())

18
vibe/cli/textual_ui/widgets/loading.py
View file

@ -15,9 +15,21 @@ from vibe.cli.textual_ui.widgets.no_markup_static import NoMarkupStatic
from vibe.cli.textual_ui.widgets.spinner import SpinnerMixin, SpinnerType
def _format_elapsed(seconds: int) -> str:
if seconds < 60: # noqa: PLR2004
return f"{seconds}s"
minutes, secs = divmod(seconds, 60)
if minutes < 60: # noqa: PLR2004
return f"{minutes}m{secs}s"
hours, mins = divmod(minutes, 60)
return f"{hours}h{mins}m{secs}s"
class LoadingWidget(SpinnerMixin, Static):
TARGET_COLORS = ("#FFD800", "#FFAF00", "#FF8205", "#FA500F", "#E10500")
SPINNER_TYPE = SpinnerType.BRAILLE
SPINNER_TYPE = SpinnerType.SNAKE
EASTER_EGGS: ClassVar[list[str]] = [
"Eating a chocolatine",
@ -173,7 +185,9 @@ class LoadingWidget(SpinnerMixin, Static):
elapsed = int(time() - self.start_time - paused)
if elapsed != self._last_elapsed:
self._last_elapsed = elapsed
self.hint_widget.update(f"({elapsed}s esc to interrupt)")
self.hint_widget.update(
f"({_format_elapsed(elapsed)} esc to interrupt)"
)
@contextmanager

73
vibe/cli/textual_ui/widgets/messages.py
View file

@ -4,9 +4,10 @@ from typing import Any
from textual.app import ComposeResult
from textual.containers import Horizontal, Vertical
from textual.widgets import Markdown, Static
from textual.widgets import Static
from textual.widgets._markdown import MarkdownStream
from vibe.cli.textual_ui.ansi_markdown import AnsiMarkdown as Markdown
from vibe.cli.textual_ui.widgets.no_markup_static import NoMarkupStatic
from vibe.cli.textual_ui.widgets.spinner import SpinnerMixin, SpinnerType
@ -42,7 +43,6 @@ class UserMessage(Static):
def compose(self) -> ComposeResult:
with Horizontal(classes="user-message-container"):
yield NonSelectableStatic("> ", classes="user-message-prompt")
yield NoMarkupStatic(self._content, classes="user-message-content")
if self._pending:
self.add_class("pending")
@ -66,6 +66,7 @@ class StreamingMessageBase(Static):
self._content = content
self._markdown: Markdown | None = None
self._stream: MarkdownStream | None = None
self._content_initialized = False
def _get_markdown(self) -> Markdown:
if self._markdown is None:
@ -89,6 +90,8 @@ class StreamingMessageBase(Static):
await stream.write(content)
async def write_initial_content(self) -> None:
if self._content_initialized:
return
if self._content and self._should_write_content():
stream = self._ensure_stream()
await stream.write(self._content)
@ -110,16 +113,15 @@ class AssistantMessage(StreamingMessageBase):
self.add_class("assistant-message")
def compose(self) -> ComposeResult:
with Horizontal(classes="assistant-message-container"):
yield NonSelectableStatic("● ", classes="assistant-message-dot")
with Vertical(classes="assistant-message-content"):
markdown = Markdown("")
self._markdown = markdown
yield markdown
if self._content:
self._content_initialized = True
markdown = Markdown(self._content)
self._markdown = markdown
yield markdown
class ReasoningMessage(SpinnerMixin, StreamingMessageBase):
SPINNER_TYPE = SpinnerType.LINE
SPINNER_TYPE = SpinnerType.PULSE
SPINNING_TEXT = "Thinking"
COMPLETED_TEXT = "Thought"
@ -207,19 +209,6 @@ class WhatsNewMessage(Static):
yield Markdown(self._content)
class PlanOfferMessage(Static):
def __init__(self, text: str) -> None:
super().__init__()
self.add_class("plan-offer-message")
self._text = text
def compose(self) -> ComposeResult:
yield Markdown(self._text)
def get_text(self) -> str:
return self._text
class InterruptMessage(Static):
def __init__(self) -> None:
super().__init__()
@ -240,30 +229,22 @@ class BashOutputMessage(Static):
self.add_class("bash-output-message")
self._command = command
self._cwd = cwd
self._output = output
self._output = output.rstrip("\n")
self._exit_code = exit_code
def compose(self) -> ComposeResult:
with Vertical(classes="bash-output-container"):
with Horizontal(classes="bash-cwd-line"):
yield NoMarkupStatic(self._cwd, classes="bash-cwd")
yield NoMarkupStatic("", classes="bash-cwd-spacer")
if self._exit_code == 0:
yield NoMarkupStatic("✓", classes="bash-exit-success")
else:
yield NoMarkupStatic("✗", classes="bash-exit-failure")
yield NoMarkupStatic(
f" ({self._exit_code})", classes="bash-exit-code"
)
with Horizontal(classes="bash-command-line"):
yield NoMarkupStatic("> ", classes="bash-chevron")
yield NoMarkupStatic(self._command, classes="bash-command")
yield NoMarkupStatic("", classes="bash-command-spacer")
status_class = "bash-success" if self._exit_code == 0 else "bash-error"
self.add_class(status_class)
with Horizontal(classes="bash-command-line"):
yield NonSelectableStatic("$ ", classes=f"bash-prompt {status_class}")
yield NoMarkupStatic(self._command, classes="bash-command")
with Horizontal(classes="bash-output-container"):
yield ExpandingBorder(classes="bash-output-border")
yield NoMarkupStatic(self._output, classes="bash-output")
class ErrorMessage(Static):
def __init__(self, error: str, collapsed: bool = True) -> None:
def __init__(self, error: str, collapsed: bool = False) -> None:
super().__init__()
self.add_class("error-message")
self._error = error
@ -274,22 +255,12 @@ class ErrorMessage(Static):
with Horizontal(classes="error-container"):
yield ExpandingBorder(classes="error-border")
self._content_widget = NoMarkupStatic(
self._get_text(), classes="error-content"
f"Error: {self._error}", classes="error-content"
)
yield self._content_widget
def _get_text(self) -> str:
if self.collapsed:
return "Error. (ctrl+o to expand)"
return f"Error: {self._error}"
def set_collapsed(self, collapsed: bool) -> None:
if self.collapsed == collapsed:
return
self.collapsed = collapsed
if self._content_widget:
self._content_widget.update(self._get_text())
pass
class WarningMessage(Static):

25
vibe/cli/textual_ui/widgets/question_app.py
View file

@ -69,26 +69,36 @@ class QuestionApp(Container):
def _current_question(self) -> Question:
return self.questions[self.current_question_idx]
@property
def _has_other(self) -> bool:
return not self._current_question.hide_other
@property
def _total_options(self) -> int:
base = len(self._current_question.options) + 1
base = len(self._current_question.options)
if self._has_other:
base += 1
if self._current_question.multi_select:
return base + 1
base += 1
return base
@property
def _other_option_idx(self) -> int:
if not self._has_other:
return -1
return len(self._current_question.options)
@property
def _submit_option_idx(self) -> int:
if not self._current_question.multi_select:
return -1
return self._other_option_idx + 1
if self._has_other:
return self._other_option_idx + 1
return len(self._current_question.options)
@property
def _is_other_selected(self) -> bool:
return self.selected_option == self._other_option_idx
return self._has_other and self.selected_option == self._other_option_idx
@property
def _is_submit_selected(self) -> bool:
@ -217,6 +227,12 @@ class QuestionApp(Container):
if not self.other_prefix or not self.other_input or not self.other_static:
return
if not self._has_other:
self.other_prefix.display = False
self.other_input.display = False
self.other_static.display = False
return
q = self._current_question
is_multi = q.multi_select
multi_selected = self.multi_selections.get(self.current_question_idx, set())
@ -235,6 +251,7 @@ class QuestionApp(Container):
show_input = is_focused or bool(stored_text)
self.other_prefix.display = True
self.other_input.display = show_input
self.other_static.display = not show_input

108
vibe/cli/textual_ui/widgets/spinner.py
View file

@ -2,11 +2,14 @@ from __future__ import annotations
from abc import ABC
from collections.abc import Callable
from enum import Enum
from enum import Enum, auto
import random
from typing import TYPE_CHECKING, Any, ClassVar, Protocol, runtime_checkable
from textual.timer import Timer
from vibe.cli.textual_ui.widgets.braille_renderer import render_braille
if TYPE_CHECKING:
from textual.widgets import Static
@ -51,47 +54,84 @@ class BrailleSpinner(Spinner):
)
class LineSpinner(Spinner):
FRAMES: ClassVar[tuple[str, ...]] = ("|", "/", "-", "\\")
class CircleSpinner(Spinner):
FRAMES: ClassVar[tuple[str, ...]] = ("◴", "◷", "◶", "◵")
class BowtieSpinner(Spinner):
class PulseSpinner(Spinner):
FRAMES: ClassVar[tuple[str, ...]] = (
"⠋",
"⠙",
"⠚",
"⠞",
"⠖",
"⠦",
"⠴",
"⠲",
"⠳",
"⠓",
"■",
"■",
"■",
"■",
"■",
"■",
"□",
"□",
"□",
"□",
)
class DotWaveSpinner(Spinner):
FRAMES: ClassVar[tuple[str, ...]] = ("⣾", "⣽", "⣻", "⢿", "⡿", "⣟", "⣯", "⣷")
class SpinnerType(Enum):
BRAILLE = "braille"
LINE = "line"
CIRCLE = "circle"
BOWTIE = "bowtie"
DOT_WAVE = "dot_wave"
BRAILLE = auto()
PULSE = auto()
SNAKE = auto()
class SnakeSpinner(Spinner):
MAP_WIDTH: ClassVar[int] = 4
MAP_HEIGHT: ClassVar[int] = 4
SNAKE_LENGTH: ClassVar[int] = 3
def __init__(self) -> None:
self._positions: list[complex] = [1, 0, 1j]
super().__init__()
@property
def current_direction(self) -> complex:
return self._positions[0] - self._positions[1]
def _is_in_bounds(self, position: complex) -> bool:
return (
0 <= position.real < self.MAP_WIDTH and 0 <= position.imag < self.MAP_HEIGHT
)
def _get_direction(self) -> complex:
if (
len(set(z.real for z in self._positions)) > 1
and len(set(z.imag for z in self._positions)) > 1
and self._is_in_bounds(self._positions[0] + self.current_direction)
):
return self.current_direction
valid_directions = []
for rotation in [1, 1j, -1j]:
offset = rotation * self.current_direction
new_position = self._positions[0] + offset
if self._is_in_bounds(new_position) and new_position not in self._positions:
valid_directions.append(offset)
return random.choice(valid_directions)
def _next_positions(self) -> list[complex]:
if len(self._positions) > self.SNAKE_LENGTH:
return self._positions[: self.SNAKE_LENGTH]
head_position = self._positions[0]
direction = self._get_direction()
if self.current_direction != direction:
return [head_position + direction] + self._positions
return [head_position + direction] + self._positions[:-1]
def current_frame(self) -> str:
return render_braille(self._positions, self.MAP_WIDTH, self.MAP_HEIGHT)
def next_frame(self) -> str:
self._positions = self._next_positions()
return self.current_frame()
def reset(self) -> None:
self._positions = [1, 0, 1j]
_SPINNER_CLASSES: dict[SpinnerType, type[Spinner]] = {
SpinnerType.BRAILLE: BrailleSpinner,
SpinnerType.LINE: LineSpinner,
SpinnerType.CIRCLE: CircleSpinner,
SpinnerType.BOWTIE: BowtieSpinner,
SpinnerType.DOT_WAVE: DotWaveSpinner,
SpinnerType.PULSE: PulseSpinner,
SpinnerType.SNAKE: SnakeSpinner,
}
@ -101,7 +141,7 @@ def create_spinner(spinner_type: SpinnerType = SpinnerType.BRAILLE) -> Spinner:
class SpinnerMixin:
SPINNER_TYPE: ClassVar[SpinnerType] = SpinnerType.LINE
SPINNER_TYPE: ClassVar[SpinnerType] = SpinnerType.BRAILLE
SPINNING_TEXT: ClassVar[str] = ""
COMPLETED_TEXT: ClassVar[str] = ""

2
vibe/cli/textual_ui/widgets/status_message.py
View file

@ -12,7 +12,7 @@ from vibe.cli.textual_ui.widgets.spinner import SpinnerMixin, SpinnerType
class StatusMessage(SpinnerMixin, NoMarkupStatic):
SPINNER_TYPE: ClassVar[SpinnerType] = SpinnerType.LINE
SPINNER_TYPE: ClassVar[SpinnerType] = SpinnerType.PULSE
def __init__(self, initial_text: str = "", **kwargs: Any) -> None:
self._initial_text = initial_text

31
vibe/cli/textual_ui/widgets/teleport_message.py Normal file
View file

@ -0,0 +1,31 @@
from __future__ import annotations
from vibe.cli.textual_ui.widgets.status_message import StatusMessage
class TeleportMessage(StatusMessage):
def __init__(self) -> None:
super().__init__()
self.add_class("teleport-message")
self._status: str = "Teleporting..."
self._final_url: str | None = None
self._error: str | None = None
def get_content(self) -> str:
if self._error:
return f"Teleport failed: {self._error}"
if self._final_url:
return f"Teleported to Nuage: {self._final_url}"
return self._status
def set_status(self, status: str) -> None:
self._status = status
self.update_display()
def set_complete(self, url: str) -> None:
self._final_url = url
self.stop_spinning(success=True)
def set_error(self, error: str) -> None:
self._error = error
self.stop_spinning(success=False)

159
vibe/cli/textual_ui/widgets/tool_widgets.py
View file

@ -6,10 +6,10 @@ from pathlib import Path
from pydantic import BaseModel
from textual.app import ComposeResult
from textual.containers import Vertical
from textual.widgets import Markdown, Static
from textual.widgets import Static
from vibe.cli.textual_ui.ansi_markdown import AnsiMarkdown as Markdown
from vibe.cli.textual_ui.widgets.no_markup_static import NoMarkupStatic
from vibe.cli.textual_ui.widgets.utils import DEFAULT_TOOL_SHORTCUT, TOOL_SHORTCUTS
from vibe.core.tools.builtins.ask_user_question import AskUserQuestionResult
from vibe.core.tools.builtins.bash import BashArgs, BashResult
from vibe.core.tools.builtins.grep import GrepArgs, GrepResult
@ -23,13 +23,13 @@ from vibe.core.tools.builtins.todo import TodoArgs, TodoResult
from vibe.core.tools.builtins.write_file import WriteFileArgs, WriteFileResult
def _truncate_lines(content: str, max_lines: int) -> str:
"""Truncate content to max_lines, adding indicator if truncated."""
def _truncate_lines(content: str, max_lines: int) -> tuple[str, str | None]:
"""Truncate content to max_lines, returning (content, truncation_info)."""
lines = content.split("\n")
if len(lines) <= max_lines:
return content
return content, None
remaining = len(lines) - max_lines
return "\n".join(lines[:max_lines] + [f"… ({remaining} more lines)"])
return "\n".join(lines[:max_lines]), f"… ({remaining} more lines)"
def parse_search_replace_to_diff(content: str) -> list[str]:
@ -90,8 +90,6 @@ class ToolApprovalWidget[TArgs: BaseModel](Vertical):
class ToolResultWidget[TResult: BaseModel](Static):
"""Base class for result widgets with typed result."""
SHORTCUT = DEFAULT_TOOL_SHORTCUT
def __init__(
self,
result: TResult | None,
@ -108,21 +106,13 @@ class ToolResultWidget[TResult: BaseModel](Static):
self.warnings = warnings or []
self.add_class("tool-result-widget")
def _hint(self) -> str:
action = "expand" if self.collapsed else "collapse"
return f"({self.SHORTCUT} to {action})"
def _header(self) -> ComposeResult:
"""Yield the standard header. Subclasses can call this then add content."""
if self.collapsed:
yield NoMarkupStatic(f"{self.message} {self._hint()}")
else:
yield NoMarkupStatic(self.message)
def _footer(self, extra: str | None = None) -> ComposeResult:
"""Yield the footer with optional extra info."""
if extra:
yield NoMarkupStatic(extra, classes="tool-result-hint")
def compose(self) -> ComposeResult:
"""Default: show message and optionally result fields."""
yield from self._header()
"""Default: show result fields."""
if not self.collapsed and self.result:
for field_name in type(self.result).model_fields:
value = getattr(self.result, field_name)
@ -130,6 +120,7 @@ class ToolResultWidget[TResult: BaseModel](Static):
yield NoMarkupStatic(
f"{field_name}: {value}", classes="tool-result-detail"
)
yield from self._footer()
class BashApprovalWidget(ToolApprovalWidget[BashArgs]):
@ -139,8 +130,17 @@ class BashApprovalWidget(ToolApprovalWidget[BashArgs]):
class BashResultWidget(ToolResultWidget[BashResult]):
def compose(self) -> ComposeResult:
yield from self._header()
if self.collapsed or not self.result:
if not self.result:
yield from self._footer()
return
if self.collapsed:
truncation_info = None
if self.result.stdout:
content, truncation_info = _truncate_lines(self.result.stdout, 10)
yield NoMarkupStatic(content, classes="tool-result-detail")
else:
yield NoMarkupStatic("(no content)", classes="tool-result-detail")
yield from self._footer(truncation_info)
return
yield NoMarkupStatic(
f"returncode: {self.result.returncode}", classes="tool-result-detail"
@ -155,6 +155,7 @@ class BashResultWidget(ToolResultWidget[BashResult]):
yield NoMarkupStatic(
f"stderr:{sep}{self.result.stderr}", classes="tool-result-detail"
)
yield from self._footer()
class WriteFileApprovalWidget(ToolApprovalWidget[WriteFileArgs]):
@ -169,8 +170,16 @@ class WriteFileApprovalWidget(ToolApprovalWidget[WriteFileArgs]):
class WriteFileResultWidget(ToolResultWidget[WriteFileResult]):
def compose(self) -> ComposeResult:
yield from self._header()
if self.collapsed or not self.result:
if not self.result:
yield from self._footer()
return
ext = Path(self.result.path).suffix.lstrip(".") or "text"
if self.collapsed:
truncation_info = None
if self.result.content:
content, truncation_info = _truncate_lines(self.result.content, 10)
yield Markdown(f"```{ext}\n{content}\n```")
yield from self._footer(truncation_info)
return
yield NoMarkupStatic(f"Path: {self.result.path}", classes="tool-result-detail")
yield NoMarkupStatic(
@ -178,8 +187,9 @@ class WriteFileResultWidget(ToolResultWidget[WriteFileResult]):
)
if self.result.content:
yield NoMarkupStatic("")
ext = Path(self.result.path).suffix.lstrip(".") or "text"
yield Markdown(f"```{ext}\n{_truncate_lines(self.result.content, 10)}\n```")
content, _ = _truncate_lines(self.result.content, 10)
yield Markdown(f"```{ext}\n{content}\n```")
yield from self._footer()
class SearchReplaceApprovalWidget(ToolApprovalWidget[SearchReplaceArgs]):
@ -196,23 +206,15 @@ class SearchReplaceApprovalWidget(ToolApprovalWidget[SearchReplaceArgs]):
class SearchReplaceResultWidget(ToolResultWidget[SearchReplaceResult]):
def compose(self) -> ComposeResult:
yield from self._header()
if self.collapsed or not self.result:
if not self.result:
yield from self._footer()
return
yield NoMarkupStatic(f"File: {self.result.file}", classes="tool-result-detail")
yield NoMarkupStatic(
f"Blocks applied: {self.result.blocks_applied}",
classes="tool-result-detail",
)
yield NoMarkupStatic(
f"Lines changed: {self.result.lines_changed}", classes="tool-result-detail"
)
for warning in self.result.warnings:
for warning in self.warnings:
yield NoMarkupStatic(f"⚠ {warning}", classes="tool-result-warning")
if self.result.content:
yield NoMarkupStatic("")
for line in parse_search_replace_to_diff(self.result.content):
yield render_diff_line(line)
yield from self._footer()
class TodoApprovalWidget(ToolApprovalWidget[TodoArgs]):
@ -227,41 +229,30 @@ class TodoApprovalWidget(ToolApprovalWidget[TodoArgs]):
class TodoResultWidget(ToolResultWidget[TodoResult]):
SHORTCUT = TOOL_SHORTCUTS["todo"]
def compose(self) -> ComposeResult:
if self.collapsed:
yield NoMarkupStatic(f"{self.message} {self._hint()}")
else:
yield NoMarkupStatic(f"{self.message} {self._hint()}")
yield NoMarkupStatic("")
if not self.result or not self.result.todos:
yield NoMarkupStatic("No todos", classes="todo-empty")
yield from self._footer()
return
if not self.result or not self.result.todos:
yield NoMarkupStatic("No todos", classes="todo-empty")
return
by_status: dict[str, list] = {
"in_progress": [],
"pending": [],
"completed": [],
"cancelled": [],
}
for todo in self.result.todos:
status = (
todo.status.value if hasattr(todo.status, "value") else str(todo.status)
)
if status in by_status:
by_status[status].append(todo)
# Group todos by status
by_status: dict[str, list] = {
"in_progress": [],
"pending": [],
"completed": [],
"cancelled": [],
}
for todo in self.result.todos:
status = (
todo.status.value
if hasattr(todo.status, "value")
else str(todo.status)
)
if status in by_status:
by_status[status].append(todo)
for status in ["in_progress", "pending", "completed", "cancelled"]:
for todo in by_status[status]:
icon = self._get_status_icon(status)
yield NoMarkupStatic(
f"{icon} {todo.content}", classes=f"todo-{status}"
)
for status in ["in_progress", "pending", "completed", "cancelled"]:
for todo in by_status[status]:
icon = self._get_status_icon(status)
yield NoMarkupStatic(f"{icon} {todo.content}", classes=f"todo-{status}")
yield from self._footer()
def _get_status_icon(self, status: str) -> str:
icons = {"pending": "☐", "in_progress": "☐", "completed": "☑", "cancelled": "☒"}
@ -283,8 +274,8 @@ class ReadFileApprovalWidget(ToolApprovalWidget[ReadFileArgs]):
class ReadFileResultWidget(ToolResultWidget[ReadFileResult]):
def compose(self) -> ComposeResult:
yield from self._header()
if self.collapsed:
yield from self._footer()
return
if self.result:
yield NoMarkupStatic(
@ -292,10 +283,13 @@ class ReadFileResultWidget(ToolResultWidget[ReadFileResult]):
)
for warning in self.warnings:
yield NoMarkupStatic(f"⚠ {warning}", classes="tool-result-warning")
truncation_info = None
if self.result and self.result.content:
yield NoMarkupStatic("")
ext = Path(self.result.path).suffix.lstrip(".") or "text"
yield Markdown(f"```{ext}\n{_truncate_lines(self.result.content, 10)}\n```")
content, truncation_info = _truncate_lines(self.result.content, 10)
yield Markdown(f"```{ext}\n{content}\n```")
yield from self._footer(truncation_info)
class GrepApprovalWidget(ToolApprovalWidget[GrepArgs]):
@ -312,20 +306,24 @@ class GrepApprovalWidget(ToolApprovalWidget[GrepArgs]):
class GrepResultWidget(ToolResultWidget[GrepResult]):
def compose(self) -> ComposeResult:
yield from self._header()
if self.collapsed:
return
for warning in self.warnings:
yield NoMarkupStatic(f"⚠ {warning}", classes="tool-result-warning")
if self.result and self.result.matches:
yield NoMarkupStatic("")
yield Markdown(f"```\n{_truncate_lines(self.result.matches, 30)}\n```")
if not self.result or not self.result.matches:
yield from self._footer()
return
max_lines = 10 if self.collapsed else None
if max_lines:
content, truncation_info = _truncate_lines(self.result.matches, max_lines)
else:
content, truncation_info = self.result.matches, None
yield NoMarkupStatic(content, classes="tool-result-detail")
yield from self._footer(truncation_info)
class AskUserQuestionResultWidget(ToolResultWidget[AskUserQuestionResult]):
def compose(self) -> ComposeResult:
if self.collapsed or not self.result:
yield from self._header()
yield from self._footer()
return
for answer in self.result.answers:
@ -333,6 +331,7 @@ class AskUserQuestionResultWidget(ToolResultWidget[AskUserQuestionResult]):
yield NoMarkupStatic(answer.question, classes="tool-result-detail")
prefix = "(Other) " if answer.is_other else ""
yield NoMarkupStatic(f"{prefix}{answer.answer}", classes="ask-user-answer")
yield from self._footer()
APPROVAL_WIDGETS: dict[str, type[ToolApprovalWidget]] = {

83
vibe/cli/textual_ui/widgets/tools.py
View file

@ -8,7 +8,6 @@ from vibe.cli.textual_ui.widgets.messages import ExpandingBorder, NonSelectableS
from vibe.cli.textual_ui.widgets.no_markup_static import NoMarkupStatic
from vibe.cli.textual_ui.widgets.status_message import StatusMessage
from vibe.cli.textual_ui.widgets.tool_widgets import get_result_widget
from vibe.cli.textual_ui.widgets.utils import DEFAULT_TOOL_SHORTCUT, TOOL_SHORTCUTS
from vibe.core.tools.ui import ToolUIDataAdapter
from vibe.core.types import ToolCallEvent, ToolResultEvent
@ -44,6 +43,14 @@ class ToolCallMessage(StatusMessage):
self._stream_widget.display = False
yield self._stream_widget
def on_mount(self) -> None:
siblings = list(self.parent.children) if self.parent else []
idx = siblings.index(self) if self in siblings else -1
if idx > 0 and isinstance(
siblings[idx - 1], (ToolCallMessage, ToolResultMessage)
):
self.add_class("no-gap")
def get_content(self) -> str:
if self._event and self._event.tool_class:
adapter = ToolUIDataAdapter(self._event.tool_class)
@ -63,6 +70,10 @@ class ToolCallMessage(StatusMessage):
self._stream_widget.display = False
super().stop_spinning(success)
def set_result_text(self, text: str) -> None:
if self._text_widget:
self._text_widget.update(text)
class ToolResultMessage(Static):
def __init__(
@ -91,13 +102,6 @@ class ToolResultMessage(Static):
def tool_name(self) -> str:
return self._tool_name
def _shortcut(self) -> str:
return TOOL_SHORTCUTS.get(self._tool_name, DEFAULT_TOOL_SHORTCUT)
def _hint(self) -> str:
action = "expand" if self.collapsed else "collapse"
return f"({self._shortcut()} to {action})"
def compose(self) -> ComposeResult:
with Horizontal(classes="tool-result-container"):
yield ExpandingBorder(classes="tool-result-border")
@ -108,6 +112,8 @@ class ToolResultMessage(Static):
if self._call_widget:
success = self._determine_success()
self._call_widget.stop_spinning(success=success)
result_text = self._get_result_text()
self._call_widget.set_result_text(result_text)
await self._render_result()
def _determine_success(self) -> bool:
@ -121,6 +127,23 @@ class ToolResultMessage(Static):
return display.success
return True
def _get_result_text(self) -> str:
if self._event is None:
return f"{self._tool_name} completed"
if self._event.error:
return f"{self._tool_name}: error"
if self._event.skipped:
return f"{self._tool_name}: skipped"
if self._event.tool_class:
adapter = ToolUIDataAdapter(self._event.tool_class)
display = adapter.get_result_display(self._event)
return display.message
return f"{self._tool_name} completed"
async def _render_result(self) -> None:
if self._content_container is None:
return
@ -128,39 +151,29 @@ class ToolResultMessage(Static):
await self._content_container.remove_children()
if self._event is None:
await self._render_simple()
self.display = False
return
if self._event.error:
self.add_class("error-text")
if self.collapsed:
await self._content_container.mount(
NoMarkupStatic(f"Error. {self._hint()}")
)
else:
await self._content_container.mount(
NoMarkupStatic(f"Error: {self._event.error}")
)
await self._content_container.mount(
NoMarkupStatic(f"Error: {self._event.error}")
)
self.display = True
return
if self._event.skipped:
self.add_class("warning-text")
reason = self._event.skip_reason or "User skipped"
if self.collapsed:
await self._content_container.mount(
NoMarkupStatic(f"Skipped. {self._hint()}")
)
else:
await self._content_container.mount(
NoMarkupStatic(f"Skipped: {reason}")
)
await self._content_container.mount(NoMarkupStatic(f"Skipped: {reason}"))
self.display = True
return
self.remove_class("error-text")
self.remove_class("warning-text")
if self._event.tool_class is None:
await self._render_simple()
self.display = False
return
adapter = ToolUIDataAdapter(self._event.tool_class)
@ -175,23 +188,7 @@ class ToolResultMessage(Static):
warnings=display.warnings,
)
await self._content_container.mount(widget)
async def _render_simple(self) -> None:
if self._content_container is None:
return
if self.collapsed:
await self._content_container.mount(
NoMarkupStatic(f"{self._tool_name} completed {self._hint()}")
)
return
if self._content:
await self._content_container.mount(NoMarkupStatic(self._content))
else:
await self._content_container.mount(
NoMarkupStatic(f"{self._tool_name} completed.")
)
self.display = bool(widget.children)
async def set_collapsed(self, collapsed: bool) -> None:
if self.collapsed == collapsed:

4
vibe/cli/textual_ui/widgets/utils.py
View file

@ -1,4 +0,0 @@
from __future__ import annotations
TOOL_SHORTCUTS = {"todo": "ctrl+t"}
DEFAULT_TOOL_SHORTCUT = "ctrl+o"

283
vibe/cli/textual_ui/widgets/welcome.py
View file

@ -1,283 +0,0 @@
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from time import monotonic
from rich.align import Align
from rich.console import Group
from rich.text import Text
from textual.color import Color
from textual.widgets import Static
from vibe import __version__
from vibe.core.config import VibeConfig
def hex_to_rgb(hex_color: str) -> tuple[int, int, int]:
normalized = hex_color.lstrip("#")
r, g, b = (int(normalized[i : i + 2], 16) for i in (0, 2, 4))
return (r, g, b)
def rgb_to_hex(r: int, g: int, b: int) -> str:
return f"#{r:02x}{g:02x}{b:02x}"
def interpolate_color(
start_rgb: tuple[int, int, int], end_rgb: tuple[int, int, int], progress: float
) -> str:
progress = max(0.0, min(1.0, progress))
r = int(start_rgb[0] + (end_rgb[0] - start_rgb[0]) * progress)
g = int(start_rgb[1] + (end_rgb[1] - start_rgb[1]) * progress)
b = int(start_rgb[2] + (end_rgb[2] - start_rgb[2]) * progress)
return rgb_to_hex(r, g, b)
@dataclass
class LineAnimationState:
progress: float = 0.0
cached_color: str | None = None
cached_progress: float = -1.0
rendered_color: str | None = None
class WelcomeBanner(Static):
FLASH_COLOR = "#FFFFFF"
TARGET_COLORS = ("#FFD800", "#FFAF00", "#FF8205", "#FA500F", "#E10500")
BORDER_TARGET_COLOR = "#b05800"
LINE_ANIMATION_DURATION_MS = 200
LINE_STAGGER_MS = 280
FLASH_RESET_DURATION_MS = 400
ANIMATION_TICK_INTERVAL = 0.1
COLOR_FLASH_MIDPOINT = 0.5
COLOR_PHASE_SCALE = 2.0
COLOR_CACHE_THRESHOLD = 0.001
BORDER_PROGRESS_THRESHOLD = 0.01
BLOCK = "▇▇"
SPACE = " "
LOGO_TEXT_GAP = " "
def __init__(self, config: VibeConfig) -> None:
super().__init__(" ")
self.config = config
self.animation_timer = None
self._animation_start_time: float | None = None
self._cached_skeleton_color: str | None = None
self._cached_skeleton_rgb: tuple[int, int, int] | None = None
self._flash_rgb = hex_to_rgb(self.FLASH_COLOR)
self._target_rgbs = [hex_to_rgb(c) for c in self.TARGET_COLORS]
self._border_target_rgb = hex_to_rgb(self.BORDER_TARGET_COLOR)
self._line_states = [LineAnimationState() for _ in self.TARGET_COLORS]
self.border_progress = 0.0
self._cached_border_color: str | None = None
self._cached_border_progress = -1.0
self._line_duration = self.LINE_ANIMATION_DURATION_MS / 1000
self._line_stagger = self.LINE_STAGGER_MS / 1000
self._border_duration = self.FLASH_RESET_DURATION_MS / 1000
self._line_start_times = [
idx * self._line_stagger for idx in range(len(self.TARGET_COLORS))
]
self._all_lines_finish_time = (
(len(self.TARGET_COLORS) - 1) * self.LINE_STAGGER_MS
+ self.LINE_ANIMATION_DURATION_MS
) / 1000
self._cached_text_lines: list[Text | None] = [None] * 7
self._initialize_static_line_suffixes()
def _initialize_static_line_suffixes(self) -> None:
self._static_line1_suffix = (
f"{self.LOGO_TEXT_GAP}[b]Mistral Vibe v{__version__}[/]"
)
self._static_line2_suffix = (
f"{self.LOGO_TEXT_GAP}[dim]{self.config.active_model}[/]"
)
mcp_count = len(self.config.mcp_servers)
model_count = len(self.config.models)
self._static_line3_suffix = f"{self.LOGO_TEXT_GAP}[dim]{model_count} models · {mcp_count} MCP servers[/]"
self._static_line5_suffix = (
f"{self.LOGO_TEXT_GAP}[dim]{self.config.displayed_workdir or Path.cwd()}[/]"
)
self._static_line7 = f"[dim]Type[/] [{self.BORDER_TARGET_COLOR}]/help[/] [dim]for more information • [/][{self.BORDER_TARGET_COLOR}]/terminal-setup[/][dim] for shift+enter[/]"
@property
def skeleton_color(self) -> str:
return self._cached_skeleton_color or "#1e1e1e"
@property
def skeleton_rgb(self) -> tuple[int, int, int]:
return self._cached_skeleton_rgb or hex_to_rgb("#1e1e1e")
def on_mount(self) -> None:
if not self.config.disable_welcome_banner_animation:
self.call_after_refresh(self._init_after_styles)
def _init_after_styles(self) -> None:
self._cache_skeleton_color()
self._cached_text_lines[5] = Text("")
self._cached_text_lines[6] = Text.from_markup(self._static_line7)
self._update_display()
self._start_animation()
def _cache_skeleton_color(self) -> None:
try:
border = self.styles.border
if (
hasattr(border, "top")
and isinstance(edge := border.top, tuple)
and len(edge) >= 2 # noqa: PLR2004
and isinstance(color := edge[1], Color)
):
self._cached_skeleton_color = color.hex
self._cached_skeleton_rgb = hex_to_rgb(color.hex)
return
except (AttributeError, TypeError):
pass
self._cached_skeleton_color = "#1e1e1e"
self._cached_skeleton_rgb = hex_to_rgb("#1e1e1e")
def _stop_timer(self) -> None:
if self.animation_timer:
try:
self.animation_timer.stop()
except Exception:
pass
self.animation_timer = None
def on_unmount(self) -> None:
self._stop_timer()
def _start_animation(self) -> None:
self._animation_start_time = monotonic()
def tick() -> None:
if self._is_animation_complete():
self._stop_timer()
return
if self._animation_start_time is None:
return
elapsed = monotonic() - self._animation_start_time
updated_lines = self._advance_line_progress(elapsed)
border_updated = self._advance_border_progress(elapsed)
if border_updated:
self._update_border_color()
if updated_lines or border_updated:
self._update_display()
self.animation_timer = self.set_interval(self.ANIMATION_TICK_INTERVAL, tick)
def _advance_line_progress(self, elapsed: float) -> bool:
any_updates = False
for line_idx, state in enumerate(self._line_states):
if state.progress >= 1.0:
continue
start_time = self._line_start_times[line_idx]
if elapsed < start_time:
continue
progress = min(1.0, (elapsed - start_time) / self._line_duration)
if progress > state.progress:
state.progress = progress
any_updates = True
return any_updates
def _advance_border_progress(self, elapsed: float) -> bool:
if elapsed < self._all_lines_finish_time:
return False
new_progress = min(
1.0, (elapsed - self._all_lines_finish_time) / self._border_duration
)
if abs(new_progress - self.border_progress) > self.BORDER_PROGRESS_THRESHOLD:
self.border_progress = new_progress
return True
return False
def _is_animation_complete(self) -> bool:
return (
all(state.progress >= 1.0 for state in self._line_states)
and self.border_progress >= 1.0
)
def _update_border_color(self) -> None:
progress = self.border_progress
if abs(progress - self._cached_border_progress) < self.COLOR_CACHE_THRESHOLD:
return
border_color = self._compute_color_for_progress(
progress, self._border_target_rgb
)
self._cached_border_color = border_color
self._cached_border_progress = progress
self.styles.border = ("round", border_color)
def _compute_color_for_progress(
self, progress: float, target_rgb: tuple[int, int, int]
) -> str:
if progress <= 0:
return self.skeleton_color
if progress <= self.COLOR_FLASH_MIDPOINT:
phase = progress * self.COLOR_PHASE_SCALE
return interpolate_color(self.skeleton_rgb, self._flash_rgb, phase)
phase = (progress - self.COLOR_FLASH_MIDPOINT) * self.COLOR_PHASE_SCALE
return interpolate_color(self._flash_rgb, target_rgb, phase)
def _update_display(self) -> None:
for idx in range(5):
self._update_colored_line(idx, idx)
lines = [line if line else Text("") for line in self._cached_text_lines]
self.update(Align.center(Group(*lines)))
def _get_color(self, line_idx: int) -> str:
state = self._line_states[line_idx]
if (
abs(state.progress - state.cached_progress) < self.COLOR_CACHE_THRESHOLD
and state.cached_color
):
return state.cached_color
color = self._compute_color_for_progress(
state.progress, self._target_rgbs[line_idx]
)
state.cached_color = color
state.cached_progress = state.progress
return color
def _update_colored_line(self, slot_idx: int, line_idx: int) -> None:
color = self._get_color(line_idx)
state = self._line_states[line_idx]
if color == state.rendered_color and self._cached_text_lines[slot_idx]:
return
state.rendered_color = color
self._cached_text_lines[slot_idx] = Text.from_markup(
self._build_line(slot_idx, color)
)
def _build_line(self, line_idx: int, color: str) -> str:
B = self.BLOCK
S = self.SPACE
patterns = [
f"{S}[{color}]{B}[/]{S}{S}{S}[{color}]{B}[/]{S}{self._static_line1_suffix}",
f"{S}[{color}]{B}{B}[/]{S}[{color}]{B}{B}[/]{S}{self._static_line2_suffix}",
f"{S}[{color}]{B}{B}{B}{B}{B}[/]{S}{self._static_line3_suffix}",
f"{S}[{color}]{B}[/]{S}[{color}]{B}[/]{S}[{color}]{B}[/]{S}",
f"[{color}]{B}{B}{B}[/]{S}[{color}]{B}{B}{B}[/]{self._static_line5_suffix}",
]
return patterns[line_idx]

29
vibe/cli/textual_ui/windowing/__init__.py Normal file
View file

@ -0,0 +1,29 @@
from __future__ import annotations
from vibe.cli.textual_ui.windowing.history import (
build_history_widgets,
non_system_history_messages,
)
from vibe.cli.textual_ui.windowing.history_windowing import (
create_resume_plan,
should_resume_history,
sync_backfill_state,
)
from vibe.cli.textual_ui.windowing.state import (
HISTORY_RESUME_TAIL_MESSAGES,
LOAD_MORE_BATCH_SIZE,
HistoryLoadMoreManager,
SessionWindowing,
)
__all__ = [
"HISTORY_RESUME_TAIL_MESSAGES",
"LOAD_MORE_BATCH_SIZE",
"HistoryLoadMoreManager",
"SessionWindowing",
"build_history_widgets",
"create_resume_plan",
"non_system_history_messages",
"should_resume_history",
"sync_backfill_state",
]

105
vibe/cli/textual_ui/windowing/history.py Normal file
View file

@ -0,0 +1,105 @@
from __future__ import annotations
from weakref import WeakKeyDictionary
from textual.widget import Widget
from vibe.cli.textual_ui.widgets.messages import (
AssistantMessage,
ReasoningMessage,
UserMessage,
)
from vibe.cli.textual_ui.widgets.tools import ToolCallMessage, ToolResultMessage
from vibe.core.types import LLMMessage, Role
def non_system_history_messages(messages: list[LLMMessage]) -> list[LLMMessage]:
return [msg for msg in messages if msg.role != Role.system]
def build_tool_call_map(messages: list[LLMMessage]) -> dict[str, str]:
tool_call_map: dict[str, str] = {}
for msg in messages:
if msg.role != Role.assistant or not msg.tool_calls:
continue
for tool_call in msg.tool_calls:
if tool_call.id:
tool_call_map[tool_call.id] = tool_call.function.name or "unknown"
return tool_call_map
def build_history_widgets(
batch: list[LLMMessage],
tool_call_map: dict[str, str],
*,
start_index: int,
tools_collapsed: bool,
history_widget_indices: WeakKeyDictionary[Widget, int],
) -> list[Widget]:
widgets: list[Widget] = []
for offset, msg in enumerate(batch):
history_index = start_index + offset
match msg.role:
case Role.user:
if msg.content:
widget = UserMessage(msg.content)
widgets.append(widget)
history_widget_indices[widget] = history_index
case Role.assistant:
if msg.content:
assistant_widget = AssistantMessage(msg.content)
widgets.append(assistant_widget)
history_widget_indices[assistant_widget] = history_index
if msg.tool_calls:
for tool_call in msg.tool_calls:
tool_name = tool_call.function.name or "unknown"
if tool_call.id:
tool_call_map[tool_call.id] = tool_name
widget = ToolCallMessage(tool_name=tool_name)
widgets.append(widget)
history_widget_indices[widget] = history_index
case Role.tool:
tool_name = msg.name or tool_call_map.get(
msg.tool_call_id or "", "tool"
)
widget = ToolResultMessage(
tool_name=tool_name, content=msg.content, collapsed=tools_collapsed
)
widgets.append(widget)
history_widget_indices[widget] = history_index
return widgets
def split_history_tail(
history_messages: list[LLMMessage], tail_size: int
) -> tuple[list[LLMMessage], list[LLMMessage], int]:
tail_messages = history_messages[-tail_size:]
backfill_messages = history_messages[:-tail_size]
tail_start_index = len(history_messages) - len(tail_messages)
return tail_messages, backfill_messages, tail_start_index
def visible_history_indices(
children: list[Widget], history_widget_indices: WeakKeyDictionary[Widget, int]
) -> list[int]:
return [
idx
for child in children
if (idx := history_widget_indices.get(child)) is not None
]
def visible_history_widgets_count(children: list[Widget]) -> int:
history_widget_types = (
UserMessage,
AssistantMessage,
ReasoningMessage,
ToolCallMessage,
ToolResultMessage,
)
return sum(isinstance(child, history_widget_types) for child in children)

71
vibe/cli/textual_ui/windowing/history_windowing.py Normal file
View file

@ -0,0 +1,71 @@
from __future__ import annotations
from dataclasses import dataclass
from weakref import WeakKeyDictionary
from textual.widget import Widget
from vibe.cli.textual_ui.widgets.messages import WhatsNewMessage
from vibe.cli.textual_ui.windowing.history import (
build_tool_call_map,
split_history_tail,
visible_history_indices,
visible_history_widgets_count,
)
from vibe.cli.textual_ui.windowing.state import SessionWindowing
from vibe.core.types import LLMMessage
@dataclass(frozen=True)
class HistoryResumePlan:
tool_call_map: dict[str, str]
tail_messages: list[LLMMessage]
backfill_messages: list[LLMMessage]
tail_start_index: int
@property
def has_backfill(self) -> bool:
return bool(self.backfill_messages)
def should_resume_history(messages_children: list[Widget]) -> bool:
allowed_pre_existing_types = (WhatsNewMessage,)
return all(
isinstance(child, allowed_pre_existing_types) for child in messages_children
)
def create_resume_plan(
history_messages: list[LLMMessage], tail_size: int
) -> HistoryResumePlan | None:
if not history_messages:
return None
tail_messages, backfill_messages, tail_start_index = split_history_tail(
history_messages, tail_size
)
return HistoryResumePlan(
tool_call_map=build_tool_call_map(history_messages),
tail_messages=tail_messages,
backfill_messages=backfill_messages,
tail_start_index=tail_start_index,
)
def sync_backfill_state(
*,
history_messages: list[LLMMessage],
messages_children: list[Widget],
history_widget_indices: WeakKeyDictionary[Widget, int],
windowing: SessionWindowing,
) -> tuple[bool, dict[str, str] | None]:
if not history_messages:
windowing.reset()
return False, None
visible_indices = visible_history_indices(messages_children, history_widget_indices)
visible_history_widgets = visible_history_widgets_count(messages_children)
has_backfill = windowing.recompute_backfill(
history_messages,
visible_indices=visible_indices,
visible_history_widgets_count=visible_history_widgets,
)
return has_backfill, build_tool_call_map(history_messages)

105
vibe/cli/textual_ui/windowing/state.py Normal file
View file

@ -0,0 +1,105 @@
from __future__ import annotations
from dataclasses import dataclass
from textual.widget import Widget
from vibe.cli.textual_ui.widgets.load_more import HistoryLoadMoreMessage
from vibe.core.types import LLMMessage
HISTORY_RESUME_TAIL_MESSAGES = 20
LOAD_MORE_BATCH_SIZE = 10
@dataclass(frozen=True)
class LoadMoreBatch:
start_index: int
messages: list[LLMMessage]
class SessionWindowing:
def __init__(self, load_more_batch_size: int) -> None:
self.load_more_batch_size = load_more_batch_size
self._backfill_messages: list[LLMMessage] = []
self._backfill_cursor = 0
@property
def remaining(self) -> int:
return self._backfill_cursor
@property
def has_backfill(self) -> bool:
return self._backfill_cursor > 0
def reset(self) -> None:
self._backfill_messages = []
self._backfill_cursor = 0
def set_backfill(self, backfill_messages: list[LLMMessage]) -> None:
self._backfill_messages = backfill_messages
self._backfill_cursor = len(backfill_messages)
def next_load_more_batch(self) -> LoadMoreBatch | None:
if self._backfill_cursor == 0:
return None
start_index = max(self._backfill_cursor - self.load_more_batch_size, 0)
batch = self._backfill_messages[start_index : self._backfill_cursor]
self._backfill_cursor = start_index
if not batch:
return None
return LoadMoreBatch(start_index=start_index, messages=batch)
def recompute_backfill(
self,
history_messages: list[LLMMessage],
visible_indices: list[int],
visible_history_widgets_count: int,
) -> bool:
if not history_messages:
self._backfill_messages = []
self._backfill_cursor = 0
return False
if visible_indices:
backfill_end = min(visible_indices)
else:
backfill_end = max(len(history_messages) - visible_history_widgets_count, 0)
self._backfill_messages = history_messages[:backfill_end]
self._backfill_cursor = len(self._backfill_messages)
return self._backfill_cursor > 0
class HistoryLoadMoreManager:
def __init__(self) -> None:
self.widget: HistoryLoadMoreMessage | None = None
async def show(self, messages_area: Widget, remaining: int) -> None:
if self.widget is None:
widget = HistoryLoadMoreMessage()
await messages_area.mount(widget, before=0)
self.widget = widget
self.set_remaining(remaining)
async def hide(self) -> None:
if self.widget is None:
return
if self.widget.parent:
await self.widget.remove()
self.widget = None
async def set_visible(
self, messages_area: Widget, *, visible: bool, remaining: int
) -> None:
if visible:
await self.show(messages_area, remaining)
return
await self.hide()
def set_enabled(self, enabled: bool) -> None:
if self.widget is None:
return
self.widget.set_enabled(enabled)
def set_remaining(self, remaining: int) -> None:
if self.widget is None:
return
self.widget.set_remaining(remaining)