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

v1.3.0

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: Thiago Padilha <thiago@coplane.com>
This commit is contained in:
Mathias Gesbert 2025-12-23 19:00:46 +01:00 • committed by Mathias Gesbert
parent 2e1e15120d
commit 078693fc64
67 changed files with 3959 additions and 819 deletions
Download patch file Download diff file Expand all files Collapse all files

2
vibe/__init__.py
View file

@ -3,4 +3,4 @@ from __future__ import annotations
from pathlib import Path
VIBE_ROOT = Path(__file__).parent
__version__ = "1.2.2"
__version__ = "1.3.0"

4
vibe/acp/acp_agent.py
View file

@ -4,7 +4,7 @@ import asyncio
from collections.abc import AsyncGenerator
from pathlib import Path
import sys
from typing import Any, cast, override
from typing import cast, override
from acp import (
PROTOCOL_VERSION,
@ -244,7 +244,7 @@ class VibeAcpAgent(AcpAgent):
return (ApprovalResponse.NO, f"Unknown option: {option_id}")
async def approval_callback(
tool_name: str, args: dict[str, Any], tool_call_id: str
tool_name: str, args: BaseModel, tool_call_id: str
) -> tuple[ApprovalResponse, str | None]:
# Create the tool call update
tool_call = ToolCall(toolCallId=tool_call_id)

2
vibe/cli/cli.py
View file

@ -26,6 +26,8 @@ def get_initial_mode(args: argparse.Namespace) -> AgentMode:
return AgentMode.PLAN
if args.auto_approve:
return AgentMode.AUTO_APPROVE
if args.prompt is not None:
return AgentMode.AUTO_APPROVE
return AgentMode.DEFAULT

50
vibe/cli/clipboard.py
View file

@ -1,7 +1,11 @@
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
@ -20,6 +24,33 @@ 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:
@ -53,18 +84,23 @@ def copy_selection_to_clipboard(app: App) -> None:
combined_text = "\n".join(selected_texts)
for copy_fn in [_copy_osc52, pyperclip.copy, app.copy_to_clipboard]:
success = False
copy_fns = _get_copy_fns(app)
for copy_fn in copy_fns:
try:
copy_fn(combined_text)
except:
pass
else:
app.notify(
f'"{_shorten_preview(selected_texts)}" copied to clipboard',
severity="information",
timeout=2,
)
break
success = True
if success:
app.notify(
f'"{_shorten_preview(selected_texts)}" copied to clipboard',
severity="information",
timeout=2,
)
else:
app.notify(
"Failed to copy - no clipboard method available",

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

@ -6,6 +6,7 @@ import subprocess
import time
from typing import Any, ClassVar, assert_never
from pydantic import BaseModel
from textual.app import App, ComposeResult
from textual.binding import Binding, BindingType
from textual.containers import Horizontal, VerticalScroll
@ -18,6 +19,10 @@ from vibe.cli.clipboard import copy_selection_to_clipboard
from vibe.cli.commands import CommandRegistry
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.approval_app import ApprovalApp
from vibe.cli.textual_ui.widgets.chat_input import ChatInputContainer
from vibe.cli.textual_ui.widgets.compact import CompactMessage
@ -29,6 +34,8 @@ from vibe.cli.textual_ui.widgets.messages import (
BashOutputMessage,
ErrorMessage,
InterruptMessage,
ReasoningMessage,
StreamingMessageBase,
UserCommandMessage,
UserMessage,
WarningMessage,
@ -116,13 +123,13 @@ class VibeApp(App):
self._mode_indicator: ModeIndicator | None = None
self._context_progress: ContextProgress | None = None
self._current_bottom_app: BottomApp = BottomApp.Input
self.theme = config.textual_theme
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._version_update_notifier = version_update_notifier
self._update_cache_repository = update_cache_repository
self._is_update_check_enabled = config.enable_update_checks
@ -138,6 +145,7 @@ class VibeApp(App):
self._agent_init_interrupted = False
self._auto_scroll = True
self._last_escape_time: float | None = None
self._terminal_theme = capture_terminal_theme()
def compose(self) -> ComposeResult:
with VerticalScroll(id="chat"):
@ -166,6 +174,15 @@ class VibeApp(App):
yield ContextProgress()
async def on_mount(self) -> None:
if self._terminal_theme:
self.register_theme(self._terminal_theme)
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.event_handler = EventHandler(
mount_callback=self._mount_and_scroll,
scroll_callback=self._scroll_to_bottom_deferred,
@ -280,7 +297,11 @@ class VibeApp(App):
def on_config_app_setting_changed(self, message: ConfigApp.SettingChanged) -> None:
if message.key == "textual_theme":
self.theme = message.value
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
@ -539,7 +560,7 @@ class VibeApp(App):
return self._agent_init_task
async def _approval_callback(
self, tool: str, args: dict, tool_call_id: str
self, tool: str, args: BaseModel, tool_call_id: str
) -> tuple[ApprovalResponse, str | None]:
self._pending_approval = asyncio.Future()
await self._switch_to_approval_app(tool, args)
@ -894,13 +915,17 @@ class VibeApp(App):
if self._mode_indicator:
self._mode_indicator.display = False
config_app = ConfigApp(self.config)
config_app = ConfigApp(
self.config, has_terminal_theme=self._terminal_theme is not None
)
await bottom_container.mount(config_app)
self._current_bottom_app = BottomApp.Config
self.call_after_refresh(config_app.focus)
async def _switch_to_approval_app(self, tool_name: str, tool_args: dict) -> None:
async def _switch_to_approval_app(
self, tool_name: str, tool_args: BaseModel
) -> None:
bottom_container = self.query_one("#bottom-app-container")
try:
@ -1133,12 +1158,36 @@ class VibeApp(App):
await self._mount_and_scroll(WarningMessage(warning, show_border=False))
async def _finalize_current_streaming_message(self) -> None:
if self._current_streaming_reasoning is not None:
self._current_streaming_reasoning.stop_spinning()
await self._current_streaming_reasoning.stop_stream()
self._current_streaming_reasoning = None
if self._current_streaming_message is None:
return
await self._current_streaming_message.stop_stream()
self._current_streaming_message = None
async def _handle_streaming_widget[T: StreamingMessageBase](
self,
widget: T,
current_stream: T | None,
other_stream: StreamingMessageBase | None,
messages_area: Widget,
) -> T | None:
if other_stream is not None:
await other_stream.stop_stream()
if current_stream is not None:
if widget._content:
await current_stream.append_content(widget._content)
return None
await messages_area.mount(widget)
await widget.write_initial_content()
return widget
async def _mount_and_scroll(self, widget: Widget) -> None:
messages_area = self.query_one("#messages")
chat = self.query_one("#chat", VerticalScroll)
@ -1147,15 +1196,28 @@ class VibeApp(App):
if was_at_bottom:
self._auto_scroll = True
if isinstance(widget, AssistantMessage):
if self._current_streaming_message is not None:
content = widget._content or ""
if content:
await self._current_streaming_message.append_content(content)
else:
self._current_streaming_message = widget
await messages_area.mount(widget)
await widget.write_initial_content()
if isinstance(widget, ReasoningMessage):
result = await self._handle_streaming_widget(
widget,
self._current_streaming_reasoning,
self._current_streaming_message,
messages_area,
)
if result is not None:
self._current_streaming_reasoning = result
self._current_streaming_message = None
elif isinstance(widget, AssistantMessage):
if self._current_streaming_reasoning is not None:
self._current_streaming_reasoning.stop_spinning()
result = await self._handle_streaming_widget(
widget,
self._current_streaming_message,
self._current_streaming_reasoning,
messages_area,
)
if result is not None:
self._current_streaming_message = result
self._current_streaming_reasoning = None
else:
await self._finalize_current_streaming_message()
await messages_area.mount(widget)
@ -1279,7 +1341,6 @@ def run_textual_ui(
initial_prompt: str | None = None,
loaded_messages: list[LLMMessage] | None = None,
) -> None:
"""Run the Textual UI."""
update_notifier = PyPIVersionUpdateGateway(project_name="mistral-vibe")
update_cache_repository = FileSystemUpdateCacheRepository()
app = VibeApp(

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

@ -227,6 +227,67 @@ Markdown MarkdownFence {
margin-bottom: 0;
}
.reasoning-message {
margin-top: 1;
width: 100%;
height: auto;
}
.reasoning-message-wrapper {
width: 100%;
height: auto;
}
.reasoning-message-header {
width: 100%;
height: auto;
}
.reasoning-indicator {
width: auto;
height: auto;
color: $text-muted;
margin-right: 1;
&.success {
color: $text-success;
}
}
.reasoning-collapsed-text {
width: auto;
height: auto;
color: $text-muted;
text-style: italic;
}
.reasoning-triangle {
width: auto;
height: auto;
color: $text-muted;
text-style: italic;
margin-left: 1;
}
.reasoning-message-content {
width: 100%;
height: auto;
padding: 0;
padding-left: 2;
margin: 0;
color: $text-muted;
text-style: italic;
}
.reasoning-message-content * {
color: $text-muted;
text-style: italic;
}
.reasoning-message-content > *:last-child {
margin-bottom: 0;
}
.interrupt-message {
margin-top: 0;
margin-bottom: 0;

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

@ -6,13 +6,14 @@ from typing import TYPE_CHECKING
from textual.widgets import Static
from vibe.cli.textual_ui.widgets.compact import CompactMessage
from vibe.cli.textual_ui.widgets.messages import AssistantMessage
from vibe.cli.textual_ui.widgets.messages import AssistantMessage, ReasoningMessage
from vibe.cli.textual_ui.widgets.tools import ToolCallMessage, ToolResultMessage
from vibe.core.types import (
AssistantEvent,
BaseEvent,
CompactEndEvent,
CompactStartEvent,
ReasoningEvent,
ToolCallEvent,
ToolResultEvent,
)
@ -50,21 +51,18 @@ class EventHandler:
return await self._handle_tool_call(event, loading_widget)
case ToolResultEvent():
sanitized_event = self._sanitize_event(event)
await self._handle_tool_result(sanitized_event)
return None
case ReasoningEvent():
await self._handle_reasoning_message(event)
case AssistantEvent():
await self._handle_assistant_message(event)
return None
case CompactStartEvent():
await self._handle_compact_start()
return None
case CompactEndEvent():
await self._handle_compact_end(event)
return None
case _:
await self._handle_unknown_event(event)
return None
return None
def _sanitize_event(self, event: ToolResultEvent) -> ToolResultEvent:
if isinstance(event, ToolResultEvent):
@ -125,6 +123,12 @@ class EventHandler:
async def _handle_assistant_message(self, event: AssistantEvent) -> None:
await self.mount_callback(AssistantMessage(event.content))
async def _handle_reasoning_message(self, event: ReasoningEvent) -> None:
tools_collapsed = self.get_tools_collapsed()
await self.mount_callback(
ReasoningMessage(event.content, collapsed=tools_collapsed)
)
async def _handle_compact_start(self) -> None:
compact_msg = CompactMessage()
self.current_compact = compact_msg

5
vibe/cli/textual_ui/renderers/__init__.py
View file

@ -1,5 +0,0 @@
from __future__ import annotations
from vibe.cli.textual_ui.renderers.tool_renderers import get_renderer
__all__ = ["get_renderer"]

216
vibe/cli/textual_ui/renderers/tool_renderers.py
View file

@ -1,216 +0,0 @@
from __future__ import annotations
import difflib
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from vibe.core.tools.ui import ToolResultDisplay
from vibe.cli.textual_ui.widgets.tool_widgets import (
BashApprovalWidget,
BashResultWidget,
GrepApprovalWidget,
GrepResultWidget,
ReadFileApprovalWidget,
ReadFileResultWidget,
SearchReplaceApprovalWidget,
SearchReplaceResultWidget,
TodoApprovalWidget,
TodoResultWidget,
ToolApprovalWidget,
ToolResultWidget,
WriteFileApprovalWidget,
WriteFileResultWidget,
)
class ToolRenderer:
def get_approval_widget(
self, tool_args: dict
) -> tuple[type[ToolApprovalWidget], dict[str, Any]]:
return ToolApprovalWidget, tool_args
def get_result_widget(
self, display: ToolResultDisplay, collapsed: bool
) -> tuple[type[ToolResultWidget], dict[str, Any]]:
data = {
"success": display.success,
"message": display.message,
"details": self._clean_details(display.details),
"warnings": display.warnings,
}
return ToolResultWidget, data
def _clean_details(self, details: dict) -> dict:
clean = {}
for key, value in details.items():
if value is None or value in ("", []):
continue
value_str = str(value).strip().replace("\n", " ").replace("\r", "")
value_str = " ".join(value_str.split())
if value_str:
clean[key] = value_str
return clean
class BashRenderer(ToolRenderer):
def get_approval_widget(
self, tool_args: dict
) -> tuple[type[BashApprovalWidget], dict[str, Any]]:
data = {
"command": tool_args.get("command", ""),
"description": tool_args.get("description", ""),
}
return BashApprovalWidget, data
def get_result_widget(
self, display: ToolResultDisplay, collapsed: bool
) -> tuple[type[BashResultWidget], dict[str, Any]]:
data = {
"success": display.success,
"message": display.message,
"details": self._clean_details(display.details),
"warnings": display.warnings,
}
return BashResultWidget, data
class WriteFileRenderer(ToolRenderer):
def get_approval_widget(
self, tool_args: dict
) -> tuple[type[WriteFileApprovalWidget], dict[str, Any]]:
data = {
"path": tool_args.get("path", ""),
"content": tool_args.get("content", ""),
"file_extension": tool_args.get("file_extension", "text"),
}
return WriteFileApprovalWidget, data
def get_result_widget(
self, display: ToolResultDisplay, collapsed: bool
) -> tuple[type[WriteFileResultWidget], dict[str, Any]]:
data = {
"success": display.success,
"message": display.message,
"path": display.details.get("path", ""),
"bytes_written": display.details.get("bytes_written"),
"content": display.details.get("content", ""),
"file_extension": display.details.get("file_extension", "text"),
}
return WriteFileResultWidget, data
class SearchReplaceRenderer(ToolRenderer):
def get_approval_widget(
self, tool_args: dict
) -> tuple[type[SearchReplaceApprovalWidget], dict[str, Any]]:
file_path = tool_args.get("file_path", "")
content = str(tool_args.get("content", ""))
diff_lines = self._parse_search_replace_blocks(content)
data = {"file_path": file_path, "diff_lines": diff_lines}
return SearchReplaceApprovalWidget, data
def get_result_widget(
self, display: ToolResultDisplay, collapsed: bool
) -> tuple[type[SearchReplaceResultWidget], dict[str, Any]]:
diff_lines = self._parse_search_replace_blocks(
display.details.get("content", "")
)
data = {
"success": display.success,
"message": display.message,
"diff_lines": diff_lines if not collapsed else [],
}
return SearchReplaceResultWidget, data
def _parse_search_replace_blocks(self, content: str) -> list[str]:
if "<<<<<<< SEARCH" not in content:
return [content]
try:
sections = content.split("<<<<<<< SEARCH")
rest = sections[1].split("=======")
search_section = rest[0].strip()
replace_part = rest[1].split(">>>>>>> REPLACE")
replace_section = replace_part[0].strip()
search_lines = search_section.split("\n")
replace_lines = replace_section.split("\n")
diff = difflib.unified_diff(search_lines, replace_lines, lineterm="", n=2)
return list(diff)[2:] # Skip file headers
except (IndexError, AttributeError):
return [content[:500]]
class TodoRenderer(ToolRenderer):
def get_approval_widget(
self, tool_args: dict
) -> tuple[type[TodoApprovalWidget], dict[str, Any]]:
data = {"description": tool_args.get("description", "")}
return TodoApprovalWidget, data
def get_result_widget(
self, display: ToolResultDisplay, collapsed: bool
) -> tuple[type[TodoResultWidget], dict[str, Any]]:
data = {
"success": display.success,
"message": display.message,
"todos_by_status": display.details.get("todos_by_status", {}),
}
return TodoResultWidget, data
class ReadFileRenderer(ToolRenderer):
def get_approval_widget(
self, tool_args: dict
) -> tuple[type[ReadFileApprovalWidget], dict[str, Any]]:
return ReadFileApprovalWidget, tool_args
def get_result_widget(
self, display: ToolResultDisplay, collapsed: bool
) -> tuple[type[ReadFileResultWidget], dict[str, Any]]:
data = {
"success": display.success,
"message": display.message,
"path": display.details.get("path", ""),
"warnings": display.warnings,
"content": display.details.get("content", "") if not collapsed else "",
"file_extension": display.details.get("file_extension", "text"),
}
return ReadFileResultWidget, data
class GrepRenderer(ToolRenderer):
def get_approval_widget(
self, tool_args: dict
) -> tuple[type[GrepApprovalWidget], dict[str, Any]]:
return GrepApprovalWidget, tool_args
def get_result_widget(
self, display: ToolResultDisplay, collapsed: bool
) -> tuple[type[GrepResultWidget], dict[str, Any]]:
data = {
"success": display.success,
"message": display.message,
"warnings": display.warnings,
"matches": display.details.get("matches", "") if not collapsed else "",
}
return GrepResultWidget, data
_RENDERER_REGISTRY: dict[str, type[ToolRenderer]] = {
"write_file": WriteFileRenderer,
"search_replace": SearchReplaceRenderer,
"todo": TodoRenderer,
"read_file": ReadFileRenderer,
"bash": BashRenderer,
"grep": GrepRenderer,
}
def get_renderer(tool_name: str) -> ToolRenderer:
renderer_class = _RENDERER_REGISTRY.get(tool_name, ToolRenderer)
return renderer_class()

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

@ -0,0 +1,266 @@
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,
)

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

@ -2,6 +2,7 @@ from __future__ import annotations
from typing import ClassVar
from pydantic import BaseModel
from textual import events
from textual.app import ComposeResult
from textual.binding import Binding, BindingType
@ -9,7 +10,7 @@ from textual.containers import Container, Vertical, VerticalScroll
from textual.message import Message
from textual.widgets import Static
from vibe.cli.textual_ui.renderers import get_renderer
from vibe.cli.textual_ui.widgets.tool_widgets import get_approval_widget
from vibe.core.config import VibeConfig
@ -29,14 +30,14 @@ class ApprovalApp(Container):
]
class ApprovalGranted(Message):
def __init__(self, tool_name: str, tool_args: dict) -> None:
def __init__(self, tool_name: str, tool_args: BaseModel) -> None:
super().__init__()
self.tool_name = tool_name
self.tool_args = tool_args
class ApprovalGrantedAlwaysTool(Message):
def __init__(
self, tool_name: str, tool_args: dict, save_permanently: bool
self, tool_name: str, tool_args: BaseModel, save_permanently: bool
) -> None:
super().__init__()
self.tool_name = tool_name
@ -44,13 +45,13 @@ class ApprovalApp(Container):
self.save_permanently = save_permanently
class ApprovalRejected(Message):
def __init__(self, tool_name: str, tool_args: dict) -> None:
def __init__(self, tool_name: str, tool_args: BaseModel) -> None:
super().__init__()
self.tool_name = tool_name
self.tool_args = tool_args
def __init__(
self, tool_name: str, tool_args: dict, workdir: str, config: VibeConfig
self, tool_name: str, tool_args: BaseModel, workdir: str, config: VibeConfig
) -> None:
super().__init__(id="approval-app")
self.tool_name = tool_name
@ -100,11 +101,8 @@ class ApprovalApp(Container):
if not self.tool_info_container:
return
renderer = get_renderer(self.tool_name)
widget_class, data = renderer.get_approval_widget(self.tool_args)
approval_widget = get_approval_widget(self.tool_name, self.tool_args)
await self.tool_info_container.remove_children()
approval_widget = widget_class(data)
await self.tool_info_container.mount(approval_widget)
def _update_options(self) -> None:

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

@ -158,7 +158,11 @@ class ChatInputContainer(Vertical):
def set_safety(self, safety: ModeSafety) -> None:
self._safety = safety
input_box = self.get_widget_by_id(self.ID_INPUT_BOX)
try:
input_box = self.get_widget_by_id(self.ID_INPUT_BOX)
except Exception:
return
for border_class in SAFETY_BORDER_CLASSES.values():
input_box.remove_class(border_class)

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

@ -10,10 +10,14 @@ 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
if TYPE_CHECKING:
from vibe.core.config import VibeConfig
THEMES = sorted(k for k in BUILTIN_THEMES if k != "textual-ansi")
_ALL_THEMES = [TERMINAL_THEME_NAME] + sorted(
k for k in BUILTIN_THEMES if k != "textual-ansi"
)
class SettingDefinition(TypedDict):
@ -46,12 +50,18 @@ class ConfigApp(Container):
super().__init__()
self.changes = changes
def __init__(self, config: VibeConfig) -> None:
def __init__(self, config: VibeConfig, *, has_terminal_theme: bool = False) -> 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",
@ -64,7 +74,7 @@ class ConfigApp(Container):
"key": "textual_theme",
"label": "Theme",
"type": "cycle",
"options": THEMES,
"options": themes,
"value": self.config.textual_theme,
},
]

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

@ -78,7 +78,7 @@ class LoadingWidget(Static):
return None
def _get_default_status(self) -> str:
return self._get_easter_egg() or "Thinking"
return self._get_easter_egg() or "Generating"
def _apply_easter_egg(self, status: str) -> str:
return self._get_easter_egg() or status

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

@ -7,6 +7,8 @@ from textual.containers import Horizontal, Vertical
from textual.widgets import Markdown, Static
from textual.widgets._markdown import MarkdownStream
from vibe.cli.textual_ui.widgets.spinner import SpinnerMixin, SpinnerType
class NonSelectableStatic(Static):
@property
@ -57,25 +59,18 @@ class UserMessage(Static):
self.remove_class("pending")
class AssistantMessage(Static):
class StreamingMessageBase(Static):
def __init__(self, content: str) -> None:
super().__init__()
self.add_class("assistant-message")
self._content = content
self._markdown: Markdown | None = None
self._stream: MarkdownStream | None = None
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
def _get_markdown(self) -> Markdown:
if self._markdown is None:
self._markdown = self.query_one(Markdown)
raise RuntimeError(
"Markdown widget not initialized. compose() must be called first."
)
return self._markdown
def _ensure_stream(self) -> MarkdownStream:
@ -88,11 +83,12 @@ class AssistantMessage(Static):
return
self._content += content
stream = self._ensure_stream()
await stream.write(content)
if self._should_write_content():
stream = self._ensure_stream()
await stream.write(content)
async def write_initial_content(self) -> None:
if self._content:
if self._content and self._should_write_content():
stream = self._ensure_stream()
await stream.write(self._content)
@ -103,6 +99,86 @@ class AssistantMessage(Static):
await self._stream.stop()
self._stream = None
def _should_write_content(self) -> bool:
return True
class AssistantMessage(StreamingMessageBase):
def __init__(self, content: str) -> None:
super().__init__(content)
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
class ReasoningMessage(SpinnerMixin, StreamingMessageBase):
SPINNER_TYPE = SpinnerType.LINE
SPINNING_TEXT = "Thinking"
COMPLETED_TEXT = "Thought"
def __init__(self, content: str, collapsed: bool = True) -> None:
super().__init__(content)
self.add_class("reasoning-message")
self.collapsed = collapsed
self._indicator_widget: Static | None = None
self._triangle_widget: Static | None = None
self.init_spinner()
def compose(self) -> ComposeResult:
with Vertical(classes="reasoning-message-wrapper"):
with Horizontal(classes="reasoning-message-header"):
self._indicator_widget = NonSelectableStatic(
self._spinner.current_frame(), classes="reasoning-indicator"
)
yield self._indicator_widget
self._status_text_widget = Static(
self.SPINNING_TEXT, markup=False, classes="reasoning-collapsed-text"
)
yield self._status_text_widget
self._triangle_widget = NonSelectableStatic(
"▶" if self.collapsed else "▼", classes="reasoning-triangle"
)
yield self._triangle_widget
markdown = Markdown("", classes="reasoning-message-content")
markdown.display = not self.collapsed
self._markdown = markdown
yield markdown
def on_mount(self) -> None:
self.start_spinner_timer()
async def on_click(self) -> None:
await self._toggle_collapsed()
async def _toggle_collapsed(self) -> None:
await self.set_collapsed(not self.collapsed)
def _should_write_content(self) -> bool:
return not self.collapsed
async def set_collapsed(self, collapsed: bool) -> None:
if self.collapsed == collapsed:
return
self.collapsed = collapsed
if self._triangle_widget:
self._triangle_widget.update("▶" if collapsed else "▼")
if self._markdown:
self._markdown.display = not collapsed
if not collapsed and self._content:
if self._stream is not None:
await self._stream.stop()
self._stream = None
await self._markdown.update("")
stream = self._ensure_stream()
await stream.write(self._content)
class UserCommandMessage(Static):
def __init__(self, content: str) -> None:

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

@ -1,8 +1,21 @@
from __future__ import annotations
from abc import ABC
from collections.abc import Callable
from enum import Enum
from typing import ClassVar
from typing import TYPE_CHECKING, Any, ClassVar, Protocol, runtime_checkable
from textual.timer import Timer
if TYPE_CHECKING:
from textual.widgets import Static
@runtime_checkable
class HasSetInterval(Protocol):
def set_interval(
self, interval: float, callback: Callable[[], None], *, name: str | None = None
) -> Timer: ...
class Spinner(ABC):
@ -85,3 +98,48 @@ _SPINNER_CLASSES: dict[SpinnerType, type[Spinner]] = {
def create_spinner(spinner_type: SpinnerType = SpinnerType.BRAILLE) -> Spinner:
spinner_class = _SPINNER_CLASSES.get(spinner_type, BrailleSpinner)
return spinner_class()
class SpinnerMixin:
SPINNER_TYPE: ClassVar[SpinnerType] = SpinnerType.LINE
SPINNING_TEXT: ClassVar[str] = ""
COMPLETED_TEXT: ClassVar[str] = ""
_spinner: Spinner
_spinner_timer: Any
_is_spinning: bool
_indicator_widget: Static | None
_status_text_widget: Static | None
def init_spinner(self) -> None:
self._spinner = create_spinner(self.SPINNER_TYPE)
self._spinner_timer = None
self._is_spinning = True
self._status_text_widget = None
def start_spinner_timer(self) -> None:
if not isinstance(self, HasSetInterval):
raise TypeError(
"SpinnerMixin requires a class that implements HasSetInterval protocol"
)
self._spinner_timer = self.set_interval(0.1, self._update_spinner_frame)
def _update_spinner_frame(self) -> None:
if not self._is_spinning or not self._indicator_widget:
return
self._indicator_widget.update(self._spinner.next_frame())
def stop_spinning(self, success: bool = True) -> None:
self._is_spinning = False
if self._spinner_timer:
self._spinner_timer.stop()
self._spinner_timer = None
if self._indicator_widget:
if success:
self._indicator_widget.update("✓")
self._indicator_widget.add_class("success")
else:
self._indicator_widget.update("✕")
self._indicator_widget.add_class("error")
if self._status_text_widget and self.COMPLETED_TEXT:
self._status_text_widget.update(self.COMPLETED_TEXT)

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

@ -7,20 +7,17 @@ from textual.containers import Horizontal
from textual.widgets import Static
from vibe.cli.textual_ui.widgets.messages import NonSelectableStatic
from vibe.cli.textual_ui.widgets.spinner import Spinner, SpinnerType, create_spinner
from vibe.cli.textual_ui.widgets.spinner import SpinnerMixin, SpinnerType
class StatusMessage(Static):
class StatusMessage(SpinnerMixin, Static):
SPINNER_TYPE: ClassVar[SpinnerType] = SpinnerType.LINE
def __init__(self, initial_text: str = "", **kwargs: Any) -> None:
self._spinner: Spinner = create_spinner(self.SPINNER_TYPE)
self._spinner_timer = None
self._is_spinning = True
self.success = True
self._initial_text = initial_text
self._indicator_widget: Static | None = None
self._text_widget: Static | None = None
self.init_spinner()
super().__init__(**kwargs)
def compose(self) -> ComposeResult:
@ -38,9 +35,9 @@ class StatusMessage(Static):
def on_mount(self) -> None:
self.update_display()
self._spinner_timer = self.set_interval(0.1, self._update_spinner)
self.start_spinner_timer()
def _update_spinner(self) -> None:
def _update_spinner_frame(self) -> None:
if not self._is_spinning:
return
self.update_display()
@ -72,4 +69,7 @@ class StatusMessage(Static):
def stop_spinning(self, success: bool = True) -> None:
self._is_spinning = False
self.success = success
if self._spinner_timer:
self._spinner_timer.stop()
self._spinner_timer = None
self.update_display()

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

@ -1,221 +1,287 @@
from __future__ import annotations
import difflib
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 vibe.cli.textual_ui.widgets.utils import DEFAULT_TOOL_SHORTCUT, TOOL_SHORTCUTS
from vibe.core.tools.builtins.bash import BashArgs, BashResult
from vibe.core.tools.builtins.grep import GrepArgs, GrepResult
from vibe.core.tools.builtins.read_file import ReadFileArgs, ReadFileResult
from vibe.core.tools.builtins.search_replace import (
SEARCH_REPLACE_BLOCK_RE,
SearchReplaceArgs,
SearchReplaceResult,
)
from vibe.core.tools.builtins.todo import TodoArgs, TodoResult
from vibe.core.tools.builtins.write_file import WriteFileArgs, WriteFileResult
class ToolApprovalWidget(Vertical):
def __init__(self, data: dict) -> None:
def _truncate_lines(content: str, max_lines: int) -> str:
"""Truncate content to max_lines, adding indicator if truncated."""
lines = content.split("\n")
if len(lines) <= max_lines:
return content
remaining = len(lines) - max_lines
return "\n".join(lines[:max_lines] + [f"… ({remaining} more lines)"])
def parse_search_replace_to_diff(content: str) -> list[str]:
"""Parse SEARCH/REPLACE blocks and generate unified diff lines."""
all_diff_lines: list[str] = []
matches = SEARCH_REPLACE_BLOCK_RE.findall(content)
if not matches:
return [content[:500]] if content else []
for i, (search_text, replace_text) in enumerate(matches):
if i > 0:
all_diff_lines.append("") # Separator between blocks
search_lines = search_text.strip().split("\n")
replace_lines = replace_text.strip().split("\n")
diff = difflib.unified_diff(search_lines, replace_lines, lineterm="", n=2)
all_diff_lines.extend(list(diff)[2:]) # Skip file headers
return all_diff_lines
def render_diff_line(line: str) -> Static:
"""Render a single diff line with appropriate styling."""
if line.startswith("---") or line.startswith("+++"):
return Static(line, markup=False, classes="diff-header")
elif line.startswith("-"):
return Static(line, markup=False, classes="diff-removed")
elif line.startswith("+"):
return Static(line, markup=False, classes="diff-added")
elif line.startswith("@@"):
return Static(line, markup=False, classes="diff-range")
else:
return Static(line, markup=False, classes="diff-context")
class ToolApprovalWidget[TArgs: BaseModel](Vertical):
"""Base class for approval widgets with typed args."""
def __init__(self, args: TArgs) -> None:
super().__init__()
self.data = data
self.args = args
self.add_class("tool-approval-widget")
def compose(self) -> ComposeResult:
MAX_APPROVAL_MSG_SIZE = 150
for key, value in self.data.items():
MAX_MSG_SIZE = 150
for field_name in type(self.args).model_fields:
value = getattr(self.args, field_name)
if value is None or value in ("", []):
continue
value_str = str(value)
if len(value_str) > MAX_APPROVAL_MSG_SIZE:
hidden = len(value_str) - MAX_APPROVAL_MSG_SIZE
value_str = (
value_str[:MAX_APPROVAL_MSG_SIZE] + f"… ({hidden} more characters)"
)
if len(value_str) > MAX_MSG_SIZE:
hidden = len(value_str) - MAX_MSG_SIZE
value_str = value_str[:MAX_MSG_SIZE] + f"… ({hidden} more characters)"
yield Static(
f"{key}: {value_str}", markup=False, classes="approval-description"
f"{field_name}: {value_str}",
markup=False,
classes="approval-description",
)
class ToolResultWidget(Static):
class ToolResultWidget[TResult: BaseModel](Static):
"""Base class for result widgets with typed result."""
SHORTCUT = DEFAULT_TOOL_SHORTCUT
def __init__(self, data: dict, collapsed: bool = True) -> None:
def __init__(
self,
result: TResult | None,
success: bool,
message: str,
collapsed: bool = True,
warnings: list[str] | None = None,
) -> None:
super().__init__()
self.data = data
self.result = result
self.success = success
self.message = message
self.collapsed = collapsed
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 compose(self) -> ComposeResult:
message = self.data.get("message", "")
def _header(self) -> ComposeResult:
"""Yield the standard header. Subclasses can call this then add content."""
if self.collapsed:
yield Static(f"{message} {self._hint()}", markup=False)
yield Static(f"{self.message} {self._hint()}", markup=False)
else:
yield Static(message, markup=False)
yield Static(self.message, markup=False)
if not self.collapsed and (details := self.data.get("details")):
for key, value in details.items():
if value:
def compose(self) -> ComposeResult:
"""Default: show message and optionally result fields."""
yield from self._header()
if not self.collapsed and self.result:
for field_name in type(self.result).model_fields:
value = getattr(self.result, field_name)
if value is not None and value not in ("", []):
yield Static(
f"{key}: {value}", markup=False, classes="tool-result-detail"
f"{field_name}: {value}",
markup=False,
classes="tool-result-detail",
)
class BashApprovalWidget(ToolApprovalWidget):
class BashApprovalWidget(ToolApprovalWidget[BashArgs]):
def compose(self) -> ComposeResult:
command = self.data.get("command", "")
description = self.data.get("description", "")
yield Markdown(f"```bash\n{self.args.command}\n```")
if description:
yield Static(description, markup=False, classes="approval-description")
class BashResultWidget(ToolResultWidget[BashResult]):
def compose(self) -> ComposeResult:
yield from self._header()
if self.collapsed or not self.result:
return
yield Static(
f"returncode: {self.result.returncode}",
markup=False,
classes="tool-result-detail",
)
if self.result.stdout:
sep = "\n" if "\n" in self.result.stdout else " "
yield Static(
f"stdout:{sep}{self.result.stdout}",
markup=False,
classes="tool-result-detail",
)
if self.result.stderr:
sep = "\n" if "\n" in self.result.stderr else " "
yield Static(
f"stderr:{sep}{self.result.stderr}",
markup=False,
classes="tool-result-detail",
)
class WriteFileApprovalWidget(ToolApprovalWidget[WriteFileArgs]):
def compose(self) -> ComposeResult:
path = Path(self.args.path)
file_extension = path.suffix.lstrip(".") or "text"
yield Static(
f"File: {self.args.path}", markup=False, classes="approval-description"
)
yield Static("")
yield Markdown(f"```{file_extension}\n{self.args.content}\n```")
class WriteFileResultWidget(ToolResultWidget[WriteFileResult]):
def compose(self) -> ComposeResult:
yield from self._header()
if self.collapsed or not self.result:
return
yield Static(
f"Path: {self.result.path}", markup=False, classes="tool-result-detail"
)
yield Static(
f"Bytes: {self.result.bytes_written}",
markup=False,
classes="tool-result-detail",
)
if self.result.content:
yield Static("")
yield Markdown(f"```bash\n{command}\n```")
ext = Path(self.result.path).suffix.lstrip(".") or "text"
yield Markdown(f"```{ext}\n{_truncate_lines(self.result.content, 10)}\n```")
class BashResultWidget(ToolResultWidget):
class SearchReplaceApprovalWidget(ToolApprovalWidget[SearchReplaceArgs]):
def compose(self) -> ComposeResult:
message = self.data.get("message", "")
if self.collapsed:
yield Static(f"{message} {self._hint()}", markup=False)
else:
yield Static(message, markup=False)
if not self.collapsed and (details := self.data.get("details")):
for key, value in details.items():
if value:
yield Static(
f"{key}: {value}", markup=False, classes="tool-result-detail"
)
class WriteFileApprovalWidget(ToolApprovalWidget):
def compose(self) -> ComposeResult:
path = self.data.get("path", "")
content = self.data.get("content", "")
file_extension = self.data.get("file_extension", "text")
yield Static(f"File: {path}", markup=False, classes="approval-description")
yield Static(
f"File: {self.args.file_path}", markup=False, classes="approval-description"
)
yield Static("")
yield Markdown(f"```{file_extension}\n{content}\n```")
diff_lines = parse_search_replace_to_diff(self.args.content)
for line in diff_lines:
yield render_diff_line(line)
class WriteFileResultWidget(ToolResultWidget):
class SearchReplaceResultWidget(ToolResultWidget[SearchReplaceResult]):
def compose(self) -> ComposeResult:
MAX_LINES = 10
message = self.data.get("message", "")
if self.collapsed:
yield Static(f"{message} {self._hint()}", markup=False)
else:
yield Static(message, markup=False)
if not self.collapsed:
if path := self.data.get("path"):
yield Static(
f"Path: {path}", markup=False, classes="tool-result-detail"
)
if bytes_written := self.data.get("bytes_written"):
yield Static(
f"Bytes: {bytes_written}",
markup=False,
classes="tool-result-detail",
)
if content := self.data.get("content"):
yield Static("")
file_extension = self.data.get("file_extension", "text")
lines = content.split("\n")
total_lines = len(lines)
if total_lines > MAX_LINES:
shown_lines = lines[:MAX_LINES]
remaining = total_lines - MAX_LINES
truncated_content = "\n".join(
shown_lines + [f"… ({remaining} more lines)"]
)
yield Markdown(f"```{file_extension}\n{truncated_content}\n```")
else:
yield Markdown(f"```{file_extension}\n{content}\n```")
class SearchReplaceApprovalWidget(ToolApprovalWidget):
def compose(self) -> ComposeResult:
file_path = self.data.get("file_path", "")
diff_lines = self.data.get("diff_lines", [])
yield Static(f"File: {file_path}", markup=False, classes="approval-description")
yield Static("")
if diff_lines:
for line in diff_lines:
if line.startswith("---") or line.startswith("+++"):
yield Static(line, markup=False, classes="diff-header")
elif line.startswith("-"):
yield Static(line, markup=False, classes="diff-removed")
elif line.startswith("+"):
yield Static(line, markup=False, classes="diff-added")
elif line.startswith("@@"):
yield Static(line, markup=False, classes="diff-range")
else:
yield Static(line, markup=False, classes="diff-context")
class SearchReplaceResultWidget(ToolResultWidget):
def compose(self) -> ComposeResult:
message = self.data.get("message", "")
if self.collapsed:
yield Static(f"{message} {self._hint()}", markup=False)
else:
yield Static(message, markup=False)
if not self.collapsed and (diff_lines := self.data.get("diff_lines")):
yield from self._header()
if self.collapsed or not self.result:
return
yield Static(
f"File: {self.result.file}", markup=False, classes="tool-result-detail"
)
yield Static(
f"Blocks applied: {self.result.blocks_applied}",
markup=False,
classes="tool-result-detail",
)
yield Static(
f"Lines changed: {self.result.lines_changed}",
markup=False,
classes="tool-result-detail",
)
for warning in self.result.warnings:
yield Static(f"⚠ {warning}", markup=False, classes="tool-result-warning")
if self.result.content:
yield Static("")
for line in diff_lines:
if line.startswith("---") or line.startswith("+++"):
yield Static(line, markup=False, classes="diff-header")
elif line.startswith("-"):
yield Static(line, markup=False, classes="diff-removed")
elif line.startswith("+"):
yield Static(line, markup=False, classes="diff-added")
elif line.startswith("@@"):
yield Static(line, markup=False, classes="diff-range")
else:
yield Static(line, markup=False, classes="diff-context")
for line in parse_search_replace_to_diff(self.result.content):
yield render_diff_line(line)
class TodoApprovalWidget(ToolApprovalWidget):
class TodoApprovalWidget(ToolApprovalWidget[TodoArgs]):
def compose(self) -> ComposeResult:
description = self.data.get("description", "")
if description:
yield Static(description, markup=False, classes="approval-description")
yield Static(
f"Action: {self.args.action}", markup=False, classes="approval-description"
)
if self.args.todos:
yield Static(
f"Todos: {len(self.args.todos)} items",
markup=False,
classes="approval-description",
)
class TodoResultWidget(ToolResultWidget):
class TodoResultWidget(ToolResultWidget[TodoResult]):
SHORTCUT = TOOL_SHORTCUTS["todo"]
def compose(self) -> ComposeResult:
message = self.data.get("message", "")
if self.collapsed:
yield Static(f"{message} {self._hint()}", markup=False)
yield Static(f"{self.message} {self._hint()}", markup=False)
else:
yield Static(f"{message} {self._hint()}", markup=False)
yield Static(f"{self.message} {self._hint()}", markup=False)
yield Static("")
by_status = self.data.get("todos_by_status", {})
if not any(by_status.values()):
if not self.result or not self.result.todos:
yield Static("No todos", markup=False, classes="todo-empty")
return
# 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"]:
todos = by_status.get(status, [])
for todo in todos:
content = todo.get("content", "")
for todo in by_status[status]:
icon = self._get_status_icon(status)
yield Static(
f"{icon} {content}", markup=False, classes=f"todo-{status}"
f"{icon} {todo.content}", markup=False, classes=f"todo-{status}"
)
def _get_status_icon(self, status: str) -> str:
@ -223,95 +289,103 @@ class TodoResultWidget(ToolResultWidget):
return icons.get(status, "☐")
class ReadFileApprovalWidget(ToolApprovalWidget):
class ReadFileApprovalWidget(ToolApprovalWidget[ReadFileArgs]):
def compose(self) -> ComposeResult:
for key, value in self.data.items():
if value:
yield Static(
f"{key}: {value}", markup=False, classes="approval-description"
)
yield Static(
f"path: {self.args.path}", markup=False, classes="approval-description"
)
if self.args.offset > 0:
yield Static(
f"offset: {self.args.offset}",
markup=False,
classes="approval-description",
)
if self.args.limit is not None:
yield Static(
f"limit: {self.args.limit}",
markup=False,
classes="approval-description",
)
class ReadFileResultWidget(ToolResultWidget):
class ReadFileResultWidget(ToolResultWidget[ReadFileResult]):
def compose(self) -> ComposeResult:
MAX_LINES = 10
message = self.data.get("message", "")
if self.collapsed:
yield Static(f"{message} {self._hint()}", markup=False)
else:
yield Static(message, markup=False)
yield from self._header()
if self.collapsed:
return
if path := self.data.get("path"):
yield Static(f"Path: {path}", markup=False, classes="tool-result-detail")
if warnings := self.data.get("warnings"):
for warning in warnings:
yield Static(
f"⚠ {warning}", markup=False, classes="tool-result-warning"
)
if content := self.data.get("content"):
if self.result:
yield Static(
f"Path: {self.result.path}", markup=False, classes="tool-result-detail"
)
for warning in self.warnings:
yield Static(f"⚠ {warning}", markup=False, classes="tool-result-warning")
if self.result and self.result.content:
yield Static("")
file_extension = self.data.get("file_extension", "text")
lines = content.split("\n")
total_lines = len(lines)
if total_lines > MAX_LINES:
shown_lines = lines[:MAX_LINES]
remaining = total_lines - MAX_LINES
truncated_content = "\n".join(
shown_lines + [f"… ({remaining} more lines)"]
)
yield Markdown(f"```{file_extension}\n{truncated_content}\n```")
else:
yield Markdown(f"```{file_extension}\n{content}\n```")
ext = Path(self.result.path).suffix.lstrip(".") or "text"
yield Markdown(f"```{ext}\n{_truncate_lines(self.result.content, 10)}\n```")
class GrepApprovalWidget(ToolApprovalWidget):
class GrepApprovalWidget(ToolApprovalWidget[GrepArgs]):
def compose(self) -> ComposeResult:
for key, value in self.data.items():
if value:
yield Static(
f"{key}: {value!s}", classes="approval-description", markup=False
)
yield Static(
f"pattern: {self.args.pattern}",
markup=False,
classes="approval-description",
)
yield Static(
f"path: {self.args.path}", markup=False, classes="approval-description"
)
if self.args.max_matches is not None:
yield Static(
f"max_matches: {self.args.max_matches}",
markup=False,
classes="approval-description",
)
class GrepResultWidget(ToolResultWidget):
class GrepResultWidget(ToolResultWidget[GrepResult]):
def compose(self) -> ComposeResult:
MAX_LINES = 30
message = self.data.get("message", "")
if self.collapsed:
yield Static(f"{message} {self._hint()}", markup=False)
else:
yield Static(message, markup=False)
yield from self._header()
if self.collapsed:
return
if warnings := self.data.get("warnings"):
for warning in warnings:
yield Static(
f"⚠ {warning}", classes="tool-result-warning", markup=False
)
if matches := self.data.get("matches"):
for warning in self.warnings:
yield Static(f"⚠ {warning}", markup=False, classes="tool-result-warning")
if self.result and self.result.matches:
yield Static("")
yield Markdown(f"```\n{_truncate_lines(self.result.matches, 30)}\n```")
lines = matches.split("\n")
total_lines = len(lines)
if total_lines > MAX_LINES:
shown_lines = lines[:MAX_LINES]
remaining = total_lines - MAX_LINES
truncated_content = "\n".join(
shown_lines + [f"… ({remaining} more lines)"]
)
yield Markdown(f"```\n{truncated_content}\n```")
else:
yield Markdown(f"```\n{matches}\n```")
APPROVAL_WIDGETS: dict[str, type[ToolApprovalWidget]] = {
"bash": BashApprovalWidget,
"read_file": ReadFileApprovalWidget,
"write_file": WriteFileApprovalWidget,
"search_replace": SearchReplaceApprovalWidget,
"grep": GrepApprovalWidget,
"todo": TodoApprovalWidget,
}
RESULT_WIDGETS: dict[str, type[ToolResultWidget]] = {
"bash": BashResultWidget,
"read_file": ReadFileResultWidget,
"write_file": WriteFileResultWidget,
"search_replace": SearchReplaceResultWidget,
"grep": GrepResultWidget,
"todo": TodoResultWidget,
}
def get_approval_widget(tool_name: str, args: BaseModel) -> ToolApprovalWidget:
widget_class = APPROVAL_WIDGETS.get(tool_name, ToolApprovalWidget)
return widget_class(args)
def get_result_widget(
tool_name: str,
result: BaseModel | None,
success: bool,
message: str,
collapsed: bool = True,
warnings: list[str] | None = None,
) -> ToolResultWidget:
widget_class = RESULT_WIDGETS.get(tool_name, ToolResultWidget)
return widget_class(result, success, message, collapsed, warnings)

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

@ -4,9 +4,9 @@ from textual.app import ComposeResult
from textual.containers import Horizontal, Vertical
from textual.widgets import Static
from vibe.cli.textual_ui.renderers import get_renderer
from vibe.cli.textual_ui.widgets.messages import ExpandingBorder
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
@ -129,11 +129,16 @@ class ToolResultMessage(Static):
adapter = ToolUIDataAdapter(self._event.tool_class)
display = adapter.get_result_display(self._event)
renderer = get_renderer(self._event.tool_name)
widget_class, data = renderer.get_result_widget(display, self.collapsed)
await self._content_container.mount(
widget_class(data, collapsed=self.collapsed)
widget = get_result_widget(
self._event.tool_name,
self._event.result,
success=display.success,
message=display.message,
collapsed=self.collapsed,
warnings=display.warnings,
)
await self._content_container.mount(widget)
async def _render_simple(self) -> None:
if self._content_container is None:

68
vibe/core/agent.py
View file

@ -4,7 +4,7 @@ import asyncio
from collections.abc import AsyncGenerator, Callable
from enum import StrEnum, auto
import time
from typing import Any, cast
from typing import cast
from uuid import uuid4
from pydantic import BaseModel
@ -28,6 +28,7 @@ from vibe.core.middleware import (
)
from vibe.core.modes import AgentMode
from vibe.core.prompts import UtilityPrompt
from vibe.core.skills.manager import SkillManager
from vibe.core.system_prompt import get_universal_system_prompt
from vibe.core.tools.base import (
BaseTool,
@ -48,6 +49,7 @@ from vibe.core.types import (
LLMChunk,
LLMMessage,
LLMUsage,
ReasoningEvent,
Role,
SyncApprovalCallback,
ToolCallEvent,
@ -103,6 +105,7 @@ class Agent:
self._max_price = max_price
self.tool_manager = ToolManager(config)
self.skill_manager = SkillManager(config)
self.format_handler = APIToolFormatHandler()
self.backend_factory = lambda: backend or self._select_backend()
@ -114,7 +117,9 @@ class Agent:
self.enable_streaming = enable_streaming
self._setup_middleware()
system_prompt = get_universal_system_prompt(self.tool_manager, config)
system_prompt = get_universal_system_prompt(
self.tool_manager, config, self.skill_manager
)
self.messages = [LLMMessage(role=Role.system, content=system_prompt)]
if self.message_observer:
@ -308,24 +313,49 @@ class Agent:
async for event in self._handle_tool_calls(resolved):
yield event
async def _stream_assistant_events(self) -> AsyncGenerator[AssistantEvent]:
batched_chunk = LLMChunk(message=LLMMessage(role=Role.assistant))
async def _stream_assistant_events(
self,
) -> AsyncGenerator[AssistantEvent | ReasoningEvent]:
content_buffer = ""
reasoning_buffer = ""
chunks_with_content = 0
chunks_with_reasoning = 0
BATCH_SIZE = 5
async for chunk in self._chat_streaming():
batched_chunk += chunk
if chunk.message.reasoning_content:
if content_buffer:
yield AssistantEvent(content=content_buffer)
content_buffer = ""
chunks_with_content = 0
reasoning_buffer += chunk.message.reasoning_content
chunks_with_reasoning += 1
if chunks_with_reasoning >= BATCH_SIZE:
yield ReasoningEvent(content=reasoning_buffer)
reasoning_buffer = ""
chunks_with_reasoning = 0
if chunk.message.content:
if reasoning_buffer:
yield ReasoningEvent(content=reasoning_buffer)
reasoning_buffer = ""
chunks_with_reasoning = 0
content_buffer += chunk.message.content
chunks_with_content += 1
if chunks_with_content >= BATCH_SIZE:
yield AssistantEvent(content=cast(str, batched_chunk.message.content))
batched_chunk = LLMChunk(message=LLMMessage(role=Role.assistant))
chunks_with_content = 0
if chunks_with_content >= BATCH_SIZE:
yield AssistantEvent(content=content_buffer)
content_buffer = ""
chunks_with_content = 0
if batched_chunk.message.content:
yield AssistantEvent(content=batched_chunk.message.content)
if reasoning_buffer:
yield ReasoningEvent(content=reasoning_buffer)
if content_buffer:
yield AssistantEvent(content=content_buffer)
async def _get_assistant_event(self) -> AssistantEvent:
llm_result = await self._chat()
@ -381,7 +411,7 @@ class Agent:
continue
decision = await self._should_execute_tool(
tool_instance, tool_call.args_dict, tool_call_id
tool_instance, tool_call.validated_args, tool_call_id
)
if decision.verdict == ToolExecutionResponse.SKIP:
@ -605,15 +635,12 @@ class Agent:
self.stats.tokens_per_second = usage.completion_tokens / time_seconds
async def _should_execute_tool(
self, tool: BaseTool, args: dict[str, Any], tool_call_id: str
self, tool: BaseTool, args: BaseModel, tool_call_id: str
) -> ToolDecision:
if self.auto_approve:
return ToolDecision(verdict=ToolExecutionResponse.EXECUTE)
args_model, _ = tool._get_tool_args_results()
validated_args = args_model.model_validate(args)
allowlist_denylist_result = tool.check_allowlist_denylist(validated_args)
allowlist_denylist_result = tool.check_allowlist_denylist(args)
if allowlist_denylist_result == ToolPermission.ALWAYS:
return ToolDecision(verdict=ToolExecutionResponse.EXECUTE)
elif allowlist_denylist_result == ToolPermission.NEVER:
@ -638,7 +665,7 @@ class Agent:
return await self._ask_approval(tool_name, args, tool_call_id)
async def _ask_approval(
self, tool_name: str, args: dict[str, Any], tool_call_id: str
self, tool_name: str, args: BaseModel, tool_call_id: str
) -> ToolDecision:
if not self.approval_callback:
return ToolDecision(
@ -844,8 +871,11 @@ class Agent:
self._max_price = max_price
self.tool_manager = ToolManager(self.config)
self.skill_manager = SkillManager(self.config)
new_system_prompt = get_universal_system_prompt(self.tool_manager, self.config)
new_system_prompt = get_universal_system_prompt(
self.tool_manager, self.config, self.skill_manager
)
self.messages = [LLMMessage(role=Role.system, content=new_system_prompt)]
if preserved_messages:

47
vibe/core/config.py
View file

@ -273,12 +273,12 @@ DEFAULT_MODELS = [
class VibeConfig(BaseSettings):
active_model: str = "devstral-2"
textual_theme: str = "terminal"
vim_keybindings: bool = False
disable_welcome_banner_animation: bool = False
displayed_workdir: str = ""
auto_compact_threshold: int = 200_000
context_warnings: bool = False
textual_theme: str = "textual-dark"
instructions: str = ""
workdir: Path | None = Field(default=None, exclude=True)
system_prompt_id: str = "cli"
@ -296,7 +296,7 @@ class VibeConfig(BaseSettings):
project_context: ProjectContextConfig = Field(default_factory=ProjectContextConfig)
session_logging: SessionLoggingConfig = Field(default_factory=SessionLoggingConfig)
tools: dict[str, BaseToolConfig] = Field(default_factory=dict)
tool_paths: list[str] = Field(
tool_paths: list[Path] = Field(
default_factory=list,
description=(
"Additional directories to search for custom tools. "
@ -326,6 +326,14 @@ class VibeConfig(BaseSettings):
),
)
skill_paths: list[Path] = Field(
default_factory=list,
description=(
"Additional directories to search for skills. "
"Each path may be absolute or relative to the current working directory."
),
)
model_config = SettingsConfigDict(
env_prefix="VIBE_", case_sensitive=False, extra="ignore"
)
@ -418,6 +426,20 @@ class VibeConfig(BaseSettings):
pass
return self
@field_validator("tool_paths", mode="before")
@classmethod
def _expand_tool_paths(cls, v: Any) -> list[Path]:
if not v:
return []
return [Path(p).expanduser().resolve() for p in v]
@field_validator("skill_paths", mode="before")
@classmethod
def _expand_skill_paths(cls, v: Any) -> list[Path]:
if not v:
return []
return [Path(p).expanduser().resolve() for p in v]
@field_validator("workdir", mode="before")
@classmethod
def _expand_workdir(cls, v: Any) -> Path | None:
@ -517,26 +539,7 @@ class VibeConfig(BaseSettings):
@classmethod
def _migrate(cls) -> None:
if not CONFIG_FILE.path.is_file():
return
try:
with CONFIG_FILE.path.open("rb") as f:
config = tomllib.load(f)
except (OSError, tomllib.TOMLDecodeError):
return
needs_save = False
if (
"auto_compact_threshold" not in config
or config["auto_compact_threshold"] == 100_000 # noqa: PLR2004
):
config["auto_compact_threshold"] = 200_000
needs_save = True
if needs_save:
cls.dump_config(config)
pass
@classmethod
def load(cls, agent: str | None = None, **overrides: Any) -> VibeConfig:

82
vibe/core/llm/backend/mistral.py
View file

@ -5,7 +5,7 @@ import json
import os
import re
import types
from typing import TYPE_CHECKING, cast
from typing import TYPE_CHECKING, NamedTuple, cast
import httpx
import mistralai
@ -27,6 +27,11 @@ if TYPE_CHECKING:
from vibe.core.config import ModelConfig, ProviderConfig
class ParsedContent(NamedTuple):
content: Content
reasoning_content: Content | None
class MistralMapper:
def prepare_message(self, msg: LLMMessage) -> mistralai.Messages:
match msg.role:
@ -35,9 +40,25 @@ class MistralMapper:
case Role.user:
return mistralai.UserMessage(role="user", content=msg.content)
case Role.assistant:
content: mistralai.AssistantMessageContent
if msg.reasoning_content:
content = [
mistralai.ThinkChunk(
type="thinking",
thinking=[
mistralai.TextChunk(
type="text", text=msg.reasoning_content
)
],
),
mistralai.TextChunk(type="text", text=msg.content or ""),
]
else:
content = msg.content or ""
return mistralai.AssistantMessage(
role="assistant",
content=msg.content,
content=content,
tool_calls=[
mistralai.ToolCall(
function=mistralai.FunctionCall(
@ -80,20 +101,37 @@ class MistralMapper:
function=mistralai.FunctionName(name=tool_choice.function.name),
)
def parse_content(self, content: mistralai.AssistantMessageContent) -> Content:
def _extract_thinking_text(self, chunk: mistralai.ThinkChunk) -> str:
thinking_content = getattr(chunk, "thinking", None)
if not thinking_content:
return ""
parts = []
for inner in thinking_content:
if hasattr(inner, "type") and inner.type == "text":
parts.append(getattr(inner, "text", ""))
elif isinstance(inner, str):
parts.append(inner)
return "".join(parts)
def parse_content(
self, content: mistralai.AssistantMessageContent
) -> ParsedContent:
if isinstance(content, str):
return content
return ParsedContent(content=content, reasoning_content=None)
concat_content = ""
concat_reasoning = ""
for chunk in content:
if isinstance(chunk, mistralai.FileChunk):
continue
match chunk.type:
case "text":
concat_content += chunk.text
case _:
pass
return concat_content
if isinstance(chunk, mistralai.TextChunk):
concat_content += chunk.text
elif isinstance(chunk, mistralai.ThinkChunk):
concat_reasoning += self._extract_thinking_text(chunk)
return ParsedContent(
content=concat_content,
reasoning_content=concat_reasoning if concat_reasoning else None,
)
def parse_tool_calls(self, tool_calls: list[mistralai.ToolCall]) -> list[ToolCall]:
return [
@ -187,14 +225,16 @@ class MistralBackend:
stream=False,
)
parsed = (
self._mapper.parse_content(response.choices[0].message.content)
if response.choices[0].message.content
else ParsedContent(content="", reasoning_content=None)
)
return LLMChunk(
message=LLMMessage(
role=Role.assistant,
content=self._mapper.parse_content(
response.choices[0].message.content
)
if response.choices[0].message.content
else "",
content=parsed.content,
reasoning_content=parsed.reasoning_content,
tool_calls=self._mapper.parse_tool_calls(
response.choices[0].message.tool_calls
)
@ -256,14 +296,16 @@ class MistralBackend:
else None,
http_headers=extra_headers,
):
parsed = (
self._mapper.parse_content(chunk.data.choices[0].delta.content)
if chunk.data.choices[0].delta.content
else ParsedContent(content="", reasoning_content=None)
)
yield LLMChunk(
message=LLMMessage(
role=Role.assistant,
content=self._mapper.parse_content(
chunk.data.choices[0].delta.content
)
if chunk.data.choices[0].delta.content
else "",
content=parsed.content,
reasoning_content=parsed.reasoning_content,
tool_calls=self._mapper.parse_tool_calls(
chunk.data.choices[0].delta.tool_calls
)

6
vibe/core/llm/format.py
View file

@ -164,7 +164,11 @@ class APIToolFormatHandler:
return "auto"
def process_api_response_message(self, message: Any) -> LLMMessage:
clean_message = {"role": message.role, "content": message.content}
clean_message = {
"role": message.role,
"content": message.content,
"reasoning_content": getattr(message, "reasoning_content", None),
}
if message.tool_calls:
clean_message["tool_calls"] = [

10
vibe/core/paths/config_paths.py
View file

@ -32,11 +32,21 @@ def _resolve_config_path(basename: str, type: Literal["file", "dir"]) -> Path:
def resolve_local_tools_dir(dir: Path) -> Path | None:
if not trusted_folders_manager.is_trusted(dir):
return None
if (candidate := dir / ".vibe" / "tools").is_dir():
return candidate
return None
def resolve_local_skills_dir(dir: Path) -> Path | None:
if not trusted_folders_manager.is_trusted(dir):
return None
if (candidate := dir / ".vibe" / "skills").is_dir():
return candidate
return None
def unlock_config_paths() -> None:
global _config_paths_locked
_config_paths_locked = False

1
vibe/core/paths/global_paths.py
View file

@ -29,6 +29,7 @@ VIBE_HOME = GlobalPath(_get_vibe_home)
GLOBAL_CONFIG_FILE = GlobalPath(lambda: VIBE_HOME.path / "config.toml")
GLOBAL_ENV_FILE = GlobalPath(lambda: VIBE_HOME.path / ".env")
GLOBAL_TOOLS_DIR = GlobalPath(lambda: VIBE_HOME.path / "tools")
GLOBAL_SKILLS_DIR = GlobalPath(lambda: VIBE_HOME.path / "skills")
SESSION_LOG_DIR = GlobalPath(lambda: VIBE_HOME.path / "logs" / "session")
TRUSTED_FOLDERS_FILE = GlobalPath(lambda: VIBE_HOME.path / "trusted_folders.toml")
LOG_DIR = GlobalPath(lambda: VIBE_HOME.path / "logs")

7
vibe/core/skills/__init__.py Normal file
View file

@ -0,0 +1,7 @@
from __future__ import annotations
from vibe.core.skills.manager import SkillManager
from vibe.core.skills.models import SkillInfo, SkillMetadata
from vibe.core.skills.parser import SkillParseError
__all__ = ["SkillInfo", "SkillManager", "SkillMetadata", "SkillParseError"]

113
vibe/core/skills/manager.py Normal file
View file

@ -0,0 +1,113 @@
from __future__ import annotations
from logging import getLogger
from pathlib import Path
from typing import TYPE_CHECKING
from vibe.core.paths.config_paths import resolve_local_skills_dir
from vibe.core.paths.global_paths import GLOBAL_SKILLS_DIR
from vibe.core.skills.models import SkillInfo, SkillMetadata
from vibe.core.skills.parser import SkillParseError, parse_frontmatter
if TYPE_CHECKING:
from vibe.core.config import VibeConfig
logger = getLogger("vibe")
class SkillManager:
def __init__(self, config: VibeConfig) -> None:
self._config = config
self._search_paths = self._compute_search_paths(config)
self.available_skills = self._discover_skills()
if self.available_skills:
logger.info(
"Discovered %d skill(s) from %d search path(s)",
len(self.available_skills),
len(self._search_paths),
)
@staticmethod
def _compute_search_paths(config: VibeConfig) -> list[Path]:
paths: list[Path] = []
for path in config.skill_paths:
if path.is_dir():
paths.append(path)
if (
skills_dir := resolve_local_skills_dir(config.effective_workdir)
) is not None:
paths.append(skills_dir)
if GLOBAL_SKILLS_DIR.path.is_dir():
paths.append(GLOBAL_SKILLS_DIR.path)
unique: list[Path] = []
for p in paths:
rp = p.resolve()
if rp not in unique:
unique.append(rp)
return unique
def _discover_skills(self) -> dict[str, SkillInfo]:
skills: dict[str, SkillInfo] = {}
for base in self._search_paths:
if not base.is_dir():
continue
for name, info in self._discover_skills_in_dir(base).items():
if name not in skills:
skills[name] = info
else:
logger.debug(
"Skipping duplicate skill '%s' at %s (already loaded from %s)",
name,
info.skill_path,
skills[name].skill_path,
)
return skills
def _discover_skills_in_dir(self, base: Path) -> dict[str, SkillInfo]:
skills: dict[str, SkillInfo] = {}
for skill_dir in base.iterdir():
if not skill_dir.is_dir():
continue
skill_file = skill_dir / "SKILL.md"
if not skill_file.is_file():
continue
if (skill_info := self._try_load_skill(skill_file)) is not None:
skills[skill_info.name] = skill_info
return skills
def _try_load_skill(self, skill_file: Path) -> SkillInfo | None:
try:
skill_info = self._parse_skill_file(skill_file)
except Exception as e:
logger.warning("Failed to parse skill at %s: %s", skill_file, e)
return None
return skill_info
def _parse_skill_file(self, skill_path: Path) -> SkillInfo:
try:
content = skill_path.read_text(encoding="utf-8")
except OSError as e:
raise SkillParseError(f"Cannot read file: {e}") from e
frontmatter, _ = parse_frontmatter(content)
metadata = SkillMetadata.model_validate(frontmatter)
skill_name_from_dir = skill_path.parent.name
if metadata.name != skill_name_from_dir:
logger.warning(
"Skill name '%s' doesn't match directory name '%s' at %s",
metadata.name,
skill_name_from_dir,
skill_path,
)
return SkillInfo.from_metadata(metadata, skill_path)
def get_skill(self, name: str) -> SkillInfo | None:
return self.available_skills.get(name)

85
vibe/core/skills/models.py Normal file
View file

@ -0,0 +1,85 @@
from __future__ import annotations
from pathlib import Path
from typing import Any
from pydantic import BaseModel, Field, field_validator
class SkillMetadata(BaseModel):
model_config = {"populate_by_name": True}
name: str = Field(
...,
min_length=1,
max_length=64,
pattern=r"^[a-z0-9]+(-[a-z0-9]+)*$",
description="Skill identifier. Lowercase letters, numbers, and hyphens only.",
)
description: str = Field(
...,
min_length=1,
max_length=1024,
description="What this skill does and when to use it.",
)
license: str | None = Field(
default=None, description="License name or reference to a bundled license file."
)
compatibility: str | None = Field(
default=None,
max_length=500,
description="Environment requirements (intended product, system packages, etc.).",
)
metadata: dict[str, str] = Field(
default_factory=dict,
description="Arbitrary key-value mapping for additional metadata.",
)
allowed_tools: list[str] = Field(
default_factory=list,
validation_alias="allowed-tools",
description="Space-delimited list of pre-approved tools (experimental).",
)
@field_validator("allowed_tools", mode="before")
@classmethod
def parse_allowed_tools(cls, v: str | list[str] | None) -> list[str]:
if v is None:
return []
if isinstance(v, str):
return v.split()
return list(v)
@field_validator("metadata", mode="before")
@classmethod
def normalize_metadata(cls, v: dict[str, Any] | None) -> dict[str, str]:
if v is None:
return {}
return {str(k): str(val) for k, val in v.items()}
class SkillInfo(BaseModel):
name: str
description: str
license: str | None = None
compatibility: str | None = None
metadata: dict[str, str] = Field(default_factory=dict)
allowed_tools: list[str] = Field(default_factory=list)
skill_path: Path
model_config = {"arbitrary_types_allowed": True}
@property
def skill_dir(self) -> Path:
return self.skill_path.parent.resolve()
@classmethod
def from_metadata(cls, meta: SkillMetadata, skill_path: Path) -> SkillInfo:
return cls(
name=meta.name,
description=meta.description,
license=meta.license,
compatibility=meta.compatibility,
metadata=meta.metadata,
allowed_tools=meta.allowed_tools,
skill_path=skill_path.resolve(),
)

39
vibe/core/skills/parser.py Normal file
View file

@ -0,0 +1,39 @@
from __future__ import annotations
import re
from typing import Any
import yaml
class SkillParseError(Exception):
def __init__(self, reason: str) -> None:
super().__init__(reason)
self.reason = reason
FM_BOUNDARY = re.compile(r"^-{3,}\s*$", re.MULTILINE)
def parse_frontmatter(content: str) -> tuple[dict[str, Any], str]:
splits = FM_BOUNDARY.split(content, 2)
if len(splits) < 3 or splits[0].strip(): # noqa: PLR2004
raise SkillParseError(
"Missing or invalid YAML frontmatter (metadata section must start and end with ---)"
)
yaml_content = splits[1]
markdown_body = splits[2]
try:
frontmatter = yaml.safe_load(yaml_content)
except yaml.YAMLError as e:
raise SkillParseError(f"Invalid YAML frontmatter: {e}") from e
if frontmatter is None:
frontmatter = {}
if not isinstance(frontmatter, dict):
raise SkillParseError("YAML frontmatter must be a mapping/dictionary")
return frontmatter, markdown_body

43
vibe/core/system_prompt.py
View file

@ -2,6 +2,7 @@ from __future__ import annotations
from collections.abc import Generator
import fnmatch
import html
import os
from pathlib import Path
import subprocess
@ -17,6 +18,7 @@ from vibe.core.utils import is_dangerous_directory, is_windows
if TYPE_CHECKING:
from vibe.core.config import ProjectContextConfig, VibeConfig
from vibe.core.skills.manager import SkillManager
from vibe.core.tools.manager import ToolManager
@ -375,7 +377,42 @@ def _add_commit_signature() -> str:
)
def get_universal_system_prompt(tool_manager: ToolManager, config: VibeConfig) -> str:
def _get_available_skills_section(skill_manager: SkillManager | None) -> str:
if skill_manager is None:
return ""
skills = skill_manager.available_skills
if not skills:
return ""
lines = [
"# Available Skills",
"",
"You have access to the following skills. When a task matches a skill's description,",
"read the full SKILL.md file to load detailed instructions.",
"",
"<available_skills>",
]
for name, info in sorted(skills.items()):
lines.append(" <skill>")
lines.append(f" <name>{html.escape(str(name))}</name>")
lines.append(
f" <description>{html.escape(str(info.description))}</description>"
)
lines.append(f" <path>{html.escape(str(info.skill_path))}</path>")
lines.append(" </skill>")
lines.append("</available_skills>")
return "\n".join(lines)
def get_universal_system_prompt(
tool_manager: ToolManager,
config: VibeConfig,
skill_manager: SkillManager | None = None,
) -> str:
sections = [config.system_prompt]
if config.include_commit_signature:
@ -398,6 +435,10 @@ def get_universal_system_prompt(tool_manager: ToolManager, config: VibeConfig) -
if user_instructions.strip():
sections.append(user_instructions)
skills_section = _get_available_skills_section(skill_manager)
if skills_section:
sections.append(skills_section)
if config.include_project_context:
is_dangerous, reason = is_dangerous_directory()
if is_dangerous:

22
vibe/core/tools/builtins/grep.py
View file

@ -284,15 +284,7 @@ class Grep(
if not event.args.use_default_ignore:
summary += " [no-ignore]"
return ToolCallDisplay(
summary=summary,
details={
"pattern": event.args.pattern,
"path": event.args.path,
"max_matches": event.args.max_matches,
"use_default_ignore": event.args.use_default_ignore,
},
)
return ToolCallDisplay(summary=summary)
@classmethod
def get_result_display(cls, event: ToolResultEvent) -> ToolResultDisplay:
@ -309,18 +301,8 @@ class Grep(
if event.result.was_truncated:
warnings.append("Output was truncated due to size/match limits")
return ToolResultDisplay(
success=True,
message=message,
warnings=warnings,
details={
"match_count": event.result.match_count,
"was_truncated": event.result.was_truncated,
"matches": event.result.matches,
},
)
return ToolResultDisplay(success=True, message=message, warnings=warnings)
@classmethod
def get_status_text(cls) -> str:
"""Return status message for spinner."""
return "Searching files"

18
vibe/core/tools/builtins/read_file.py
View file

@ -188,14 +188,7 @@ class ReadFile(
parts.append(f"limit {event.args.limit} lines")
summary += f" ({', '.join(parts)})"
return ToolCallDisplay(
summary=summary,
details={
"path": event.args.path,
"offset": event.args.offset,
"limit": event.args.limit,
},
)
return ToolCallDisplay(summary=summary)
@classmethod
def get_result_display(cls, event: ToolResultEvent) -> ToolResultDisplay:
@ -215,15 +208,6 @@ class ReadFile(
warnings=["File was truncated due to size limit"]
if event.result.was_truncated
else [],
details={
"path": str(event.result.path),
"lines_read": event.result.lines_read,
"was_truncated": event.result.was_truncated,
"content": event.result.content,
"file_extension": path_obj.suffix.lstrip(".")
if path_obj.suffix
else "text",
},
)
@classmethod

17
vibe/core/tools/builtins/search_replace.py
View file

@ -13,11 +13,11 @@ from vibe.core.tools.base import BaseTool, BaseToolConfig, BaseToolState, ToolEr
from vibe.core.tools.ui import ToolCallDisplay, ToolResultDisplay, ToolUIData
from vibe.core.types import ToolCallEvent, ToolResultEvent
_BLOCK_RE = re.compile(
SEARCH_REPLACE_BLOCK_RE = re.compile(
r"<{5,} SEARCH\r?\n(.*?)\r?\n?={5,}\r?\n(.*?)\r?\n?>{5,} REPLACE", flags=re.DOTALL
)
_BLOCK_WITH_FENCE_RE = re.compile(
SEARCH_REPLACE_BLOCK_WITH_FENCE_RE = re.compile(
r"```[\s\S]*?\n<{5,} SEARCH\r?\n(.*?)\r?\n?={5,}\r?\n(.*?)\r?\n?>{5,} REPLACE\s*\n```",
flags=re.DOTALL,
)
@ -88,11 +88,6 @@ class SearchReplace(
return ToolCallDisplay(
summary=f"Patching {args.file_path} ({len(blocks)} blocks)",
content=args.content,
details={
"path": args.file_path,
"blocks_count": len(blocks),
"original_path": args.file_path,
},
)
@classmethod
@ -102,10 +97,6 @@ class SearchReplace(
success=True,
message=f"Applied {event.result.blocks_applied} block{'' if event.result.blocks_applied == 1 else 's'}",
warnings=event.result.warnings,
details={
"lines_changed": event.result.lines_changed,
"content": event.result.content,
},
)
return ToolResultDisplay(success=True, message="Patch applied")
@ -406,10 +397,10 @@ class SearchReplace(
1. With code block fences (```...```)
2. Without code block fences
"""
matches = _BLOCK_WITH_FENCE_RE.findall(content)
matches = SEARCH_REPLACE_BLOCK_WITH_FENCE_RE.findall(content)
if not matches:
matches = _BLOCK_RE.findall(content)
matches = SEARCH_REPLACE_BLOCK_RE.findall(content)
return [
SearchReplaceBlock(

25
vibe/core/tools/builtins/todo.py
View file

@ -75,20 +75,12 @@ class Todo(
match args.action:
case "read":
return ToolCallDisplay(
summary="Reading todos", details={"action": "read"}
)
return ToolCallDisplay(summary="Reading todos")
case "write":
count = len(args.todos) if args.todos else 0
return ToolCallDisplay(
summary=f"Writing {count} todos",
details={"action": "write", "count": count},
)
return ToolCallDisplay(summary=f"Writing {count} todos")
case _:
return ToolCallDisplay(
summary=f"Unknown action: {args.action}",
details={"action": args.action},
)
return ToolCallDisplay(summary=f"Unknown action: {args.action}")
@classmethod
def get_result_display(cls, event: ToolResultEvent) -> ToolResultDisplay:
@ -97,16 +89,7 @@ class Todo(
result = event.result
by_status = {"in_progress": [], "pending": [], "completed": [], "cancelled": []}
for todo in result.todos:
by_status[todo.status].append({"content": todo.content, "id": todo.id})
return ToolResultDisplay(
success=True,
message=result.message,
details={"todos_by_status": by_status, "total_count": result.total_count},
)
return ToolResultDisplay(success=True, message=result.message)
@classmethod
def get_status_text(cls) -> str:

15
vibe/core/tools/builtins/write_file.py
View file

@ -56,17 +56,10 @@ class WriteFile(
return ToolCallDisplay(summary="Invalid arguments")
args = event.args
file_ext = Path(args.path).suffix.lstrip(".")
return ToolCallDisplay(
summary=f"Writing {args.path}{' (overwrite)' if args.overwrite else ''}",
content=args.content,
details={
"path": args.path,
"overwrite": args.overwrite,
"file_extension": file_ext,
"content": args.content,
},
)
@classmethod
@ -74,13 +67,7 @@ class WriteFile(
if isinstance(event.result, WriteFileResult):
action = "Overwritten" if event.result.file_existed else "Created"
return ToolResultDisplay(
success=True,
message=f"{action} {Path(event.result.path).name}",
details={
"bytes_written": event.result.bytes_written,
"path": event.result.path,
"content": event.result.content,
},
success=True, message=f"{action} {Path(event.result.path).name}"
)
return ToolResultDisplay(success=True, message="File written")

11
vibe/core/tools/manager.py
View file

@ -19,7 +19,6 @@ from vibe.core.tools.mcp import (
list_tools_http,
list_tools_stdio,
)
from vibe.core.trusted_folders import trusted_folders_manager
from vibe.core.utils import run_sync
logger = getLogger("vibe")
@ -53,17 +52,11 @@ class ToolManager:
def _compute_search_paths(config: VibeConfig) -> list[Path]:
paths: list[Path] = [DEFAULT_TOOL_DIR.path]
for p in config.tool_paths:
path = Path(p).expanduser().resolve()
for path in config.tool_paths:
if path.is_dir():
paths.append(path)
is_folder_trusted = trusted_folders_manager.is_trusted(config.effective_workdir)
if (
is_folder_trusted is True
and (tools_dir := resolve_local_tools_dir(config.effective_workdir))
is not None
):
if (tools_dir := resolve_local_tools_dir(config.effective_workdir)) is not None:
paths.append(tools_dir)
if GLOBAL_TOOLS_DIR.path.is_dir():

34
vibe/core/tools/mcp.py
View file

@ -173,12 +173,7 @@ def create_mcp_http_proxy_tool_class(
@classmethod
def get_call_display(cls, event: ToolCallEvent) -> ToolCallDisplay:
return ToolCallDisplay(
summary=f"{published_name}",
details=event.args.model_dump()
if hasattr(event.args, "model_dump")
else {},
)
return ToolCallDisplay(summary=f"{published_name}")
@classmethod
def get_result_display(cls, event: ToolResultEvent) -> ToolResultDisplay:
@ -189,15 +184,7 @@ def create_mcp_http_proxy_tool_class(
)
message = f"MCP tool {event.result.tool} completed"
details = {}
if event.result.text:
details["text"] = event.result.text
if event.result.structured:
details["structured"] = event.result.structured
return ToolResultDisplay(
success=event.result.ok, message=message, details=details
)
return ToolResultDisplay(success=event.result.ok, message=message)
@classmethod
def get_status_text(cls) -> str:
@ -279,12 +266,7 @@ def create_mcp_stdio_proxy_tool_class(
@classmethod
def get_call_display(cls, event: ToolCallEvent) -> ToolCallDisplay:
return ToolCallDisplay(
summary=f"{published_name}",
details=event.args.model_dump()
if hasattr(event.args, "model_dump")
else {},
)
return ToolCallDisplay(summary=f"{published_name}")
@classmethod
def get_result_display(cls, event: ToolResultEvent) -> ToolResultDisplay:
@ -295,15 +277,7 @@ def create_mcp_stdio_proxy_tool_class(
)
message = f"MCP tool {event.result.tool} completed"
details = {}
if event.result.text:
details["text"] = event.result.text
if event.result.structured:
details["structured"] = event.result.structured
return ToolResultDisplay(
success=event.result.ok, message=message, details=details
)
return ToolResultDisplay(success=event.result.ok, message=message)
@classmethod
def get_status_text(cls) -> str:

16
vibe/core/tools/ui.py
View file

@ -11,14 +11,12 @@ if TYPE_CHECKING:
class ToolCallDisplay(BaseModel):
summary: str # Brief description: "Writing file.txt", "Patching code.py"
content: str | None = None # Optional content preview
details: dict[str, Any] = Field(default_factory=dict) # Tool-specific data
class ToolResultDisplay(BaseModel):
success: bool
message: str
warnings: list[str] = Field(default_factory=list)
details: dict[str, Any] = Field(default_factory=dict) # Tool-specific data
@runtime_checkable
@ -46,9 +44,7 @@ class ToolUIDataAdapter:
args_dict = event.args.model_dump() if hasattr(event.args, "model_dump") else {}
args_str = ", ".join(f"{k}={v!r}" for k, v in list(args_dict.items())[:3])
return ToolCallDisplay(
summary=f"{event.tool_name}({args_str})", details=args_dict
)
return ToolCallDisplay(summary=f"{event.tool_name}({args_str})")
def get_result_display(self, event: ToolResultEvent) -> ToolResultDisplay:
if event.error:
@ -62,15 +58,7 @@ class ToolUIDataAdapter:
if self.ui_data_class:
return self.ui_data_class.get_result_display(event)
return ToolResultDisplay(
success=True,
message="Success",
details=(
event.result.model_dump()
if event.result and hasattr(event.result, "model_dump")
else {}
),
)
return ToolResultDisplay(success=True, message="Success")
def get_status_text(self) -> str:
if self.ui_data_class:

21
vibe/core/types.py
View file

@ -168,6 +168,7 @@ class LLMMessage(BaseModel):
role: Role
content: Content | None = None
reasoning_content: Content | None = None
tool_calls: list[ToolCall] | None = None
name: str | None = None
tool_call_id: str | None = None
@ -178,10 +179,15 @@ class LLMMessage(BaseModel):
if isinstance(v, dict):
v.setdefault("content", "")
v.setdefault("role", "assistant")
v.setdefault(
"reasoning_content", v.get("reasoning_content") or v.get("reasoning")
)
return v
return {
"role": str(getattr(v, "role", "assistant")),
"content": getattr(v, "content", ""),
"reasoning_content": getattr(v, "reasoning_content", None)
or getattr(v, "reasoning", None),
"tool_calls": getattr(v, "tool_calls", None),
"name": getattr(v, "name", None),
"tool_call_id": getattr(v, "tool_call_id", None),
@ -202,6 +208,12 @@ class LLMMessage(BaseModel):
if not content:
content = None
reasoning_content = (self.reasoning_content or "") + (
other.reasoning_content or ""
)
if not reasoning_content:
reasoning_content = None
tool_calls_map = OrderedDict[int, ToolCall]()
for tool_calls in [self.tool_calls or [], other.tool_calls or []]:
for tc in tool_calls:
@ -226,6 +238,7 @@ class LLMMessage(BaseModel):
return LLMMessage(
role=self.role,
content=content,
reasoning_content=reasoning_content,
tool_calls=list(tool_calls_map.values()) or None,
name=self.name,
tool_call_id=self.tool_call_id,
@ -275,6 +288,10 @@ class AssistantEvent(BaseEvent):
)
class ReasoningEvent(BaseEvent):
content: str
class ToolCallEvent(BaseEvent):
tool_name: str
tool_class: type[BaseTool]
@ -311,11 +328,11 @@ class OutputFormat(StrEnum):
type AsyncApprovalCallback = Callable[
[str, dict[str, Any], str], Awaitable[tuple[ApprovalResponse, str | None]]
[str, BaseModel, str], Awaitable[tuple[ApprovalResponse, str | None]]
]
type SyncApprovalCallback = Callable[
[str, dict[str, Any], str], tuple[ApprovalResponse, str | None]
[str, BaseModel, str], tuple[ApprovalResponse, str | None]
]
type ApprovalCallback = AsyncApprovalCallback | SyncApprovalCallback

13
vibe/setup/onboarding/__init__.py
View file

@ -4,7 +4,12 @@ import sys
from rich import print as rprint
from textual.app import App
from textual.theme import Theme
from vibe.cli.textual_ui.terminal_theme import (
TERMINAL_THEME_NAME,
capture_terminal_theme,
)
from vibe.core.paths.global_paths import GLOBAL_ENV_FILE
from vibe.setup.onboarding.screens import (
ApiKeyScreen,
@ -16,7 +21,15 @@ from vibe.setup.onboarding.screens import (
class OnboardingApp(App[str | None]):
CSS_PATH = "onboarding.tcss"
def __init__(self) -> None:
super().__init__()
self._terminal_theme: Theme | None = capture_terminal_theme()
def on_mount(self) -> None:
if self._terminal_theme:
self.register_theme(self._terminal_theme)
self.theme = TERMINAL_THEME_NAME
self.install_screen(WelcomeScreen(), "welcome")
self.install_screen(ThemeSelectionScreen(), "theme_selection")
self.install_screen(ApiKeyScreen(), "api_key")

43
vibe/setup/onboarding/screens/theme_selection.py
View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import ClassVar
from typing import TYPE_CHECKING, ClassVar
from textual.app import ComposeResult
from textual.binding import Binding, BindingType
@ -9,10 +9,16 @@ from textual.events import Resize
from textual.theme import BUILTIN_THEMES
from textual.widgets import Markdown, Static
from vibe.cli.textual_ui.terminal_theme import TERMINAL_THEME_NAME
from vibe.core.config import VibeConfig
from vibe.setup.onboarding.base import OnboardingScreen
THEMES = sorted(k for k in BUILTIN_THEMES if k != "textual-ansi")
if TYPE_CHECKING:
from vibe.setup.onboarding import OnboardingApp
THEMES = [TERMINAL_THEME_NAME] + sorted(
k for k in BUILTIN_THEMES if k != "textual-ansi"
)
VISIBLE_NEIGHBORS = 3
FADE_CLASSES = ["fade-1", "fade-2", "fade-3"]
@ -73,7 +79,7 @@ class ThemeSelectionScreen(OnboardingScreen):
Horizontal(
Static("Navigate ↑ ↓", id="nav-hint"),
Vertical(*self._compose_theme_list(), id="theme-list"),
Static("Press Enter ↵", id="enter-hint"),
Static("Press Enter \u21b5", id="enter-hint"),
id="theme-row",
)
)
@ -83,10 +89,24 @@ class ThemeSelectionScreen(OnboardingScreen):
with preview:
yield Container(Markdown(PREVIEW_MARKDOWN), id="preview-inner")
@property
def _has_terminal_theme(self) -> bool:
app: OnboardingApp = self.app # type: ignore[assignment]
return app._terminal_theme is not None
@property
def _available_themes(self) -> list[str]:
if self._has_terminal_theme:
return THEMES
return [t for t in THEMES if t != TERMINAL_THEME_NAME]
def on_mount(self) -> None:
current_theme = self.app.theme
if current_theme in THEMES:
self._theme_index = THEMES.index(current_theme)
themes = self._available_themes
if current_theme == TERMINAL_THEME_NAME:
self._theme_index = 0
elif current_theme in themes:
self._theme_index = themes.index(current_theme)
self._update_display()
self._update_preview_height()
self.focus()
@ -102,8 +122,9 @@ class ThemeSelectionScreen(OnboardingScreen):
preview.styles.max_height = max(10, available)
def _get_theme_at_offset(self, offset: int) -> str:
index = (self._theme_index + offset) % len(THEMES)
return THEMES[index]
themes = self._available_themes
index = (self._theme_index + offset) % len(themes)
return themes[index]
def _update_display(self) -> None:
for i, widget in enumerate(self._theme_widgets):
@ -121,8 +142,10 @@ class ThemeSelectionScreen(OnboardingScreen):
widget.add_class(FADE_CLASSES[distance])
def _navigate(self, direction: int) -> None:
self._theme_index = (self._theme_index + direction) % len(THEMES)
self.app.theme = THEMES[self._theme_index]
themes = self._available_themes
self._theme_index = (self._theme_index + direction) % len(themes)
theme = themes[self._theme_index]
self.app.theme = theme
self._update_display()
def action_next_theme(self) -> None:
@ -132,7 +155,7 @@ class ThemeSelectionScreen(OnboardingScreen):
self._navigate(-1)
def action_next(self) -> None:
theme = THEMES[self._theme_index]
theme = self._available_themes[self._theme_index]
try:
VibeConfig.save_updates({"textual_theme": theme})
except OSError:

24
vibe/setup/trusted_folders/trust_folder_dialog.py
View file

@ -11,6 +11,10 @@ from textual.containers import CenterMiddle, Horizontal
from textual.message import Message
from textual.widgets import Static
from vibe.cli.textual_ui.terminal_theme import (
TERMINAL_THEME_NAME,
capture_terminal_theme,
)
from vibe.core.paths.global_paths import GLOBAL_CONFIG_FILE, TRUSTED_FOLDERS_FILE
@ -145,9 +149,13 @@ class TrustFolderApp(App):
self.folder_path = folder_path
self._result: bool | None = None
self._quit_without_saving = False
self._terminal_theme = capture_terminal_theme()
self._load_theme()
def _load_theme(self) -> None:
if self._terminal_theme:
self.register_theme(self._terminal_theme)
config_file = GLOBAL_CONFIG_FILE.path
if not config_file.is_file():
return
@ -155,10 +163,18 @@ class TrustFolderApp(App):
try:
with config_file.open("rb") as f:
config_data = tomllib.load(f)
if textual_theme := config_data.get("textual_theme"):
self.theme = textual_theme
except (OSError, tomllib.TOMLDecodeError, KeyError):
pass
except (OSError, tomllib.TOMLDecodeError):
return
textual_theme = config_data.get("textual_theme")
if not textual_theme:
return
if textual_theme == TERMINAL_THEME_NAME:
if self._terminal_theme:
self.theme = TERMINAL_THEME_NAME
else:
self.theme = textual_theme
def compose(self) -> ComposeResult:
yield TrustFolderDialog(self.folder_path)