Merge branch 'feature/header-panel-agent-modes'

This commit is contained in:
2026-03-11 21:36:27 -05:00
6 changed files with 178 additions and 21 deletions

View File

@@ -7,7 +7,7 @@ import time
from typing import TYPE_CHECKING, Any
from app.agent.context import SessionContext
from app.models.config import AppConfig
from app.models.config import AgentMode, AppConfig
from app.models.message import Message
from app.models.tool_call import ToolCall, ToolResult, ToolResultStatus
from app.services.llm import LLMClient, LLMConnectionError, LLMError, LLMStreamError
@@ -59,6 +59,12 @@ class AgentLoop:
self._skills = skills_manager
self._skill_runner = skill_runner
self._tools_schema = registry.get_openai_tools_schema()
if self._permissions.mode == AgentMode.PLAN:
read_only = PermissionsService.READ_ONLY_TOOLS
self._tools_schema = [
t for t in self._tools_schema
if t["function"]["name"] in read_only
]
self._system_prompt = self._build_system_prompt()
self._cancelled = False
@@ -89,6 +95,14 @@ class AgentLoop:
f"\n\nCurrently active skill: {self._skill_runner.active_skill_name}. "
"When the skill's objective is complete, call the `finish_skill` tool."
)
if self._permissions.mode == AgentMode.PLAN:
prompt += (
"\n\nYou are in PLAN mode. You may only use read-only tools: "
"read_file, list_dir, grep_files, find_files, finish. "
"Do NOT attempt to write files, edit code, or run commands. "
"Instead, describe what changes you would make, which files "
"you would modify, and provide the reasoning for each change."
)
return prompt
def _get_messages_with_system_prompt(self) -> list[Message]:

View File

@@ -1,6 +1,7 @@
"""Pydantic configuration models mapping to config/config.yaml."""
import os
from enum import StrEnum
from pathlib import Path
from typing import Any
@@ -8,6 +9,14 @@ import yaml
from pydantic import BaseModel, Field, model_validator
class AgentMode(StrEnum):
"""Runtime agent mode controlling permission behavior."""
NORMAL = "normal"
PLAN = "plan"
AUTO = "auto"
class LLMConfig(BaseModel):
"""LLM backend configuration."""

View File

@@ -8,7 +8,7 @@ import re
import shlex
from collections.abc import Awaitable, Callable
from app.models.config import PermissionsConfig, ToolsConfig
from app.models.config import AgentMode, PermissionsConfig, ToolsConfig
logger = logging.getLogger(__name__)
@@ -30,6 +30,10 @@ class PermissionsService:
shows a modal dialog. Without a callback, unlisted tools are denied.
"""
READ_ONLY_TOOLS: frozenset[str] = frozenset({
"read_file", "list_dir", "grep_files", "find_files", "finish",
})
def __init__(
self,
config: PermissionsConfig,
@@ -38,6 +42,16 @@ class PermissionsService:
self.config = config
self._tools_config = tools_config
self._prompt_callback: PromptCallback | None = None
self._mode: AgentMode = AgentMode.NORMAL
@property
def mode(self) -> AgentMode:
"""Current agent mode."""
return self._mode
@mode.setter
def mode(self, value: AgentMode) -> None:
self._mode = value
def set_prompt_callback(self, callback: PromptCallback) -> None:
"""Set the async callback used to prompt the user for permission.
@@ -67,6 +81,16 @@ class PermissionsService:
logger.info("Tool '%s' is in deny list — blocked", tool_name)
return False
if self._mode == AgentMode.AUTO:
logger.debug("Tool '%s' auto-approved (AUTO mode)", tool_name)
return True
if self._mode == AgentMode.PLAN:
if tool_name not in self.READ_ONLY_TOOLS:
logger.info("Tool '%s' blocked in Plan mode (read-only tools only)", tool_name)
return False
return True
if tool_name in self.config.auto_approve:
logger.debug("Tool '%s' is auto-approved", tool_name)
return True

View File

@@ -10,18 +10,19 @@ from rich.panel import Panel
from rich.text import Text
from textual.app import App, ComposeResult
from textual.binding import Binding
from textual.widgets import Header, Input, RichLog
from textual.widgets import Input, RichLog
from textual import work
from app.agent.context import SessionContext
from app.agent.loop import AgentLoop
from app.models.config import AppConfig
from app.models.config import AgentMode, AppConfig
from app.services.llm import LLMClient
from app.services.permissions import PermissionsService
from app.services.session import SessionManager
from app.services.streaming import StreamHandler
from app.tools.registry import create_default_registry
from app.ui.widgets import (
HeaderPanel,
HistoryInput,
PermissionModal,
SessionResumeModal,
@@ -45,6 +46,7 @@ class SneakyCodeApp(App):
BINDINGS = [
Binding("ctrl+c", "cancel_or_quit", "Cancel/Quit", show=False),
Binding("ctrl+p", "cycle_mode", "Cycle Mode"),
]
def __init__(self, config: AppConfig, session_mgr: SessionManager | None = None) -> None:
@@ -61,10 +63,9 @@ class SneakyCodeApp(App):
self._skill_runner = None
self._current_worker: Worker | None = None
self._cancel_count = 0
self.sub_title = config.llm.model
def compose(self) -> ComposeResult:
yield Header()
yield HeaderPanel(model_name=self._config.llm.model)
yield RichLog(id="chat-log", highlight=True, markup=True)
yield StreamingStatic("", id="streaming")
yield StatusBar()
@@ -190,6 +191,8 @@ class SneakyCodeApp(App):
table.add_row("/session", "Show session info (messages, tokens, start time)")
table.add_row("/models", "List available Ollama models")
table.add_row("/models <name>", "Switch to a different model")
table.add_row("/mode", "Show current agent mode")
table.add_row("/mode normal|plan|auto", "Switch agent mode")
table.add_row("/skills", "List available skills")
table.add_row("/<skill>", "Load a skill by name")
log.write(table)
@@ -243,8 +246,23 @@ class SneakyCodeApp(App):
else:
new_model = parts[1].strip()
self._config.llm.model = new_model
self.sub_title = new_model
self.query_one(HeaderPanel).update_model(new_model)
log.write(Text(f"Switched to model: {new_model}", style="bold green"))
elif cmd.startswith("/mode"):
parts = command.split(maxsplit=1)
if len(parts) == 1:
current = self._permissions.mode
log.write(Text(f"Current mode: {current.value}", style="cyan"))
else:
mode_str = parts[1].strip().lower()
try:
new_mode = AgentMode(mode_str)
except ValueError:
log.write(Text(f"Unknown mode: {mode_str}. Use normal, plan, or auto.", style="yellow"))
return
self._permissions.mode = new_mode
self.query_one(HeaderPanel).update_mode(new_mode)
log.write(Text(f"Switched to {new_mode.value} mode", style="bold green"))
elif cmd == "/skills":
if self._skills_manager:
skills = self._skills_manager.list_skills()
@@ -306,12 +324,19 @@ class SneakyCodeApp(App):
status_bar.start_streaming()
# Set up streaming UI callbacks
header = self.query_one(HeaderPanel)
def on_content(content: str) -> None:
streaming_widget.update(
Panel(Markdown(content), title="Assistant", border_style="green", expand=True)
)
streaming_widget.show_streaming()
status_bar.update_stream_tokens(len(content) // 4)
stream_tokens = len(content) // 4
status_bar.update_stream_tokens(stream_tokens)
header.update_tokens(
self._ctx.estimated_tokens + stream_tokens,
self._ctx.token_counter.budget,
)
def on_thinking() -> None:
streaming_widget.update(Text("Thinking...", style="dim"))
@@ -335,6 +360,10 @@ class SneakyCodeApp(App):
status_bar.stop_streaming()
# Update token display in header
header = self.query_one(HeaderPanel)
header.update_tokens(self._ctx.estimated_tokens, self._ctx.token_counter.budget)
# Update skill indicator (skill may have been deactivated via finish_skill)
if self._skill_runner and not self._skill_runner.is_active:
status_bar.set_active_skill(None)
@@ -360,6 +389,21 @@ class SneakyCodeApp(App):
log = self.query_one("#chat-log", RichLog)
log.write(Text("⚠ Cancelling... (press Ctrl+C again to quit)", style="yellow"))
def action_cycle_mode(self) -> None:
"""Cycle through agent modes: Normal → Plan → Auto → Normal."""
if self._permissions is None:
return
cycle = {
AgentMode.NORMAL: AgentMode.PLAN,
AgentMode.PLAN: AgentMode.AUTO,
AgentMode.AUTO: AgentMode.NORMAL,
}
new_mode = cycle[self._permissions.mode]
self._permissions.mode = new_mode
self.query_one(HeaderPanel).update_mode(new_mode)
log = self.query_one("#chat-log", RichLog)
log.write(Text(f"Mode: {new_mode.value}", style="bold green"))
async def on_unmount(self) -> None:
"""Clean up the LLM client on app shutdown."""
if self._client is not None:

View File

@@ -55,8 +55,8 @@ Screen {
Input {
dock: bottom;
margin: 0;
border: none;
padding: 0 1;
}
Header {
dock: top;
}
/* HeaderPanel styles are in DEFAULT_CSS on the widget itself */

View File

@@ -11,6 +11,82 @@ from textual.widgets import Button, Input, Static
from rich.text import Text
from app.models.config import AgentMode
# ---------------------------------------------------------------------------
# Header Panel
# ---------------------------------------------------------------------------
class HeaderPanel(Static):
"""Single-line header showing model name, agent mode, and token usage."""
DEFAULT_CSS = """
HeaderPanel {
dock: top;
height: 1;
background: $accent;
color: $text;
padding: 0 2;
}
"""
def __init__(self, model_name: str) -> None:
super().__init__("")
self._model_name = model_name
self._mode: AgentMode = AgentMode.NORMAL
self._tokens: int = 0
self._budget: int = 0
def on_resize(self) -> None:
self._refresh_display()
def update_model(self, name: str) -> None:
"""Update the displayed model name."""
self._model_name = name
self._refresh_display()
def update_mode(self, mode: AgentMode) -> None:
"""Update the displayed agent mode."""
self._mode = mode
self._refresh_display()
def update_tokens(self, tokens: int, budget: int) -> None:
"""Update the token usage display."""
self._tokens = tokens
self._budget = budget
self._refresh_display()
def _refresh_display(self) -> None:
"""Rebuild the header text."""
left = Text.assemble(
("⚡ SneakyCode", "bold"),
"",
(self._model_name, "bold"),
)
mode_styles = {
AgentMode.NORMAL: ("NORMAL", "bold black on white"),
AgentMode.PLAN: ("PLAN", "bold black on yellow"),
AgentMode.AUTO: ("AUTO", "bold white on red"),
}
mode_label, mode_style = mode_styles[self._mode]
mode_text = Text.assemble((" ", mode_style), (mode_label, mode_style), (" ", mode_style))
right = Text(f"~{self._tokens:,} / {self._budget:,} tokens")
# Pad between sections
total_content = left.plain + " " + mode_text.plain + " " + right.plain
available = self.size.width if self.size.width > 0 else 80
gap_left = max(1, (available - len(total_content)) // 2)
gap_right = max(1, available - len(total_content) - gap_left)
full = Text.assemble(
left, " " * gap_left, mode_text, " " * gap_right, right,
)
self.update(full)
# ---------------------------------------------------------------------------
# Modal Dialogs
@@ -139,8 +215,6 @@ class StatusBar(Static):
def __init__(self) -> None:
super().__init__("")
self._tokens: int = 0
self._budget: int = 0
self._iteration: int = 0
self._max_iterations: int = 0
self._streaming: bool = False
@@ -149,12 +223,6 @@ class StatusBar(Static):
self._stream_tokens: int = 0
self._active_skill: str | None = None
def update_tokens(self, tokens: int, budget: int) -> None:
"""Update the token usage display."""
self._tokens = tokens
self._budget = budget
self._refresh_display()
def update_iteration(self, iteration: int, max_iterations: int) -> None:
"""Update the iteration count display."""
self._iteration = iteration
@@ -200,8 +268,6 @@ class StatusBar(Static):
parts.append(f"{spinner} Thinking")
if self._stream_tokens > 0:
parts.append(f"~{self._stream_tokens:,} tokens")
if self._budget > 0:
parts.append(f"Tokens: ~{self._tokens:,} / {self._budget:,}")
if self._max_iterations > 0:
parts.append(f"Iteration {self._iteration}/{self._max_iterations}")
self.update(Text(" \u2502 ".join(parts), style="dim"))