sneakygeek/SneakyCode
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'feature/tweaks-implementation'

Browse Source
This commit is contained in:
Phillip Tarrant
2026-03-11 15:48:10 -05:00
parent 7600195ecf 3f9012e6c2
commit f93b28215d
13 changed files with 683 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
app/agent/loop.py
View File

@@ -1,7 +1,10 @@
"""AgentLoop — ReAct-style tool-call loop for autonomous task execution."""
from __future__ import annotations
import json
from typing import Any
import time
from typing import TYPE_CHECKING, Any
from app.agent.context import SessionContext
from app.models.config import AppConfig
@@ -14,6 +17,10 @@ from app.tools.registry import ToolRegistry
from app.utils.display import DisplayAdapter
from app.utils.logging import get_logger
if TYPE_CHECKING:
from app.services.debug_log import DebugLogger
from app.services.skills import SkillsManager
logger = get_logger(__name__)
_MAX_REASONING_RETRIES = 2
@@ -36,6 +43,8 @@ class AgentLoop:
registry: ToolRegistry,
permissions: PermissionsService,
display: DisplayAdapter | None = None,
debug_logger: DebugLogger | None = None,
skills_manager: SkillsManager | None = None,
) -> None:
self._config = config
self._ctx = ctx
@@ -44,6 +53,8 @@ class AgentLoop:
self._registry = registry
self._permissions = permissions
self._display = display
self._debug = debug_logger
self._skills = skills_manager
self._tools_schema = registry.get_openai_tools_schema()
self._system_prompt = self._build_system_prompt()
self._cancelled = False
@@ -55,7 +66,7 @@ class AgentLoop:
def _build_system_prompt(self) -> str:
"""Build the system prompt including tool schemas and agent instructions."""
tool_names = [t["function"]["name"] for t in self._tools_schema]
return (
prompt = (
"You are SneakyCode, a local AI coding agent. "
"You help users with software engineering tasks by reading files, "
"searching code, and answering questions about their project.\n\n"
@@ -68,6 +79,9 @@ class AgentLoop:
"with a brief summary. If you can answer directly without tools, just respond "
"with text (no tool call needed)."
)
if self._skills:
prompt += self._skills.get_system_prompt_snippet()
return prompt
def _get_messages_with_system_prompt(self) -> list[Message]:
"""Prepend the system prompt to conversation history."""
@@ -195,9 +209,16 @@ class AgentLoop:
The assistant Message, or None if an error occurred.
"""
messages = self._get_messages_with_system_prompt()
if self._debug:
self._debug.log_request(messages, self._config.llm.model)
t0 = time.monotonic()
try:
chunk_iter = self._client.stream_chat_with_retry(messages, tools=self._tools_schema)
return await self._handler.process_stream(chunk_iter)
result = await self._handler.process_stream(chunk_iter)
if result and self._debug:
elapsed = (time.monotonic() - t0) * 1000
self._debug.log_response(result, self._handler.usage, elapsed)
return result
except KeyboardInterrupt:
if self._display:
self._display.write_warning("Response interrupted.")
@@ -268,7 +289,7 @@ class AgentLoop:
# Check permissions (truncate args for display in prompt)
desc = tc.function.arguments[:120] + "..." if len(tc.function.arguments) > 120 else tc.function.arguments
if not await self._permissions.check(name, description=desc):
if not await self._permissions.check(name, description=desc, arguments=tc.function.arguments):
result = ToolResult(
tool_call_id=tc_id,
tool_name=name,
@@ -281,9 +302,14 @@ class AgentLoop:
continue
# Execute tool (BaseTool.run never raises)
tool_t0 = time.monotonic()
result = tool.run(tc_id, parsed_args)
results.append(result)
if self._debug:
tool_elapsed = (time.monotonic() - tool_t0) * 1000
self._debug.log_tool_execution(name, result.status.value, tool_elapsed)
if self._config.display.show_tool_calls and self._display:
is_error = result.status == ToolResultStatus.ERROR
output = result.error if is_error else result.output

15
app/main.py
View File

@@ -36,6 +36,13 @@ def parse_args() -> argparse.Namespace:
default=None,
help="Path to log file for persistent logging",
)
parser.add_argument(
"directory",
nargs="?",
type=Path,
default=None,
help="Project directory to use as workspace root (default: current directory)",
)
return parser.parse_args()
@@ -63,6 +70,14 @@ def main() -> None:
print_error(f"Configuration error: {e}")
sys.exit(1)
# Override workspace root if directory argument provided
if args.directory:
target = Path(args.directory).resolve()
if not target.is_dir():
print_error(f"Not a directory: {target}")
sys.exit(1)
config.agent.workspace_root = target
logger.info("config_loaded", model=config.llm.model, endpoint=config.llm.endpoint)
# Pre-TUI startup info (printed to console before Textual takes over)

20
app/models/config.py
View File

@@ -90,6 +90,24 @@ class DisplayConfig(BaseModel):
stream_output: bool = Field(default=True, description="Stream LLM output to terminal")
class SkillsConfig(BaseModel):
"""Skills system configuration."""
enabled: bool = Field(default=True, description="Enable skills system")
directories: list[Path] = Field(
default_factory=lambda: [Path(".sneakycode/skills")],
description="Directories to scan for skill markdown files",
)
class DebugConfig(BaseModel):
"""Debug logging configuration."""
enabled: bool = Field(default=False, description="Enable debug logging")
log_dir: Path = Field(default=Path(".sneakycode/logs"), description="Debug log directory")
max_files: int = Field(default=10, description="Max debug log files to retain")
class AppConfig(BaseModel):
"""Top-level application configuration composing all sub-configs."""
@@ -99,6 +117,8 @@ class AppConfig(BaseModel):
tools: ToolsConfig = Field(default_factory=ToolsConfig)
display: DisplayConfig = Field(default_factory=DisplayConfig)
session: SessionConfig = Field(default_factory=SessionConfig)
debug: DebugConfig = Field(default_factory=DebugConfig)
skills: SkillsConfig = Field(default_factory=SkillsConfig)
@model_validator(mode="after")
def resolve_workspace_root(self) -> "AppConfig":

77
app/services/debug_log.py Normal file
View File

@@ -0,0 +1,77 @@
"""Debug logger — writes detailed LLM interaction logs to JSONL files."""
from __future__ import annotations
import json
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from app.models.message import Message
class DebugLogger:
"""Writes detailed LLM interaction logs to JSONL files for debugging."""
def __init__(self, log_dir: Path, max_files: int = 10) -> None:
self._log_dir = log_dir
self._log_dir.mkdir(parents=True, exist_ok=True)
self._max_files = max_files
self._file = self._log_dir / f"debug_{datetime.now(UTC).strftime('%Y%m%d_%H%M%S')}.jsonl"
self._rotate()
def log_request(self, messages: list[Message], model: str) -> None:
"""Log outbound LLM request (message roles/lengths, model)."""
self._write({
"event": "llm_request",
"model": model,
"message_count": len(messages),
"messages": [
{"role": m.role, "content_len": len(m.content or "")}
for m in messages
],
})
def log_response(
self,
message: Message,
usage: Any | None,
elapsed_ms: float,
) -> None:
"""Log LLM response with timing and token counts."""
self._write({
"event": "llm_response",
"elapsed_ms": round(elapsed_ms, 1),
"content_len": len(message.content or ""),
"tool_call_count": len(message.tool_calls or []),
"tool_calls": [tc.function.name for tc in (message.tool_calls or [])],
"usage": usage.__dict__ if usage else None,
})
def log_tool_execution(
self,
tool_name: str,
result_status: str,
elapsed_ms: float,
) -> None:
"""Log tool execution with timing."""
self._write({
"event": "tool_execution",
"tool": tool_name,
"status": result_status,
"elapsed_ms": round(elapsed_ms, 1),
})
def _write(self, record: dict[str, Any]) -> None:
record["timestamp"] = datetime.now(UTC).isoformat()
with open(self._file, "a") as f:
f.write(json.dumps(record) + "\n")
def _rotate(self) -> None:
"""Remove old debug log files beyond max_files."""
files = sorted(
self._log_dir.glob("debug_*.jsonl"),
key=lambda p: p.stat().st_mtime,
)
while len(files) > self._max_files:
files.pop(0).unlink()

19
app/services/llm.py
View File

@@ -60,6 +60,25 @@ class LLMClient:
timeout=httpx.Timeout(config.timeout, connect=10.0),
)
async def list_models(self) -> list[dict[str, str]]:
"""Query Ollama /api/tags for available models.
Returns:
List of dicts with 'name' and 'size' keys.
Raises:
LLMConnectionError: If the endpoint is unreachable.
"""
try:
response = await self._client.get("/api/tags")
data = response.json()
return [
{"name": m.get("name", ""), "size": str(m.get("size", ""))}
for m in data.get("models", [])
]
except (httpx.HTTPError, httpx.TimeoutException) as e:
raise LLMConnectionError(f"Failed to list models: {e}") from e
async def preflight_check(self) -> None:
"""Verify the endpoint is reachable and the configured model is available.

67
app/services/permissions.py
View File

@@ -2,10 +2,12 @@
from __future__ import annotations
import json
import logging
import shlex
from collections.abc import Awaitable, Callable
from app.models.config import PermissionsConfig
from app.models.config import PermissionsConfig, ToolsConfig
logger = logging.getLogger(__name__)
@@ -24,8 +26,13 @@ class PermissionsService:
shows a modal dialog. Without a callback, unlisted tools are denied.
"""
def __init__(self, config: PermissionsConfig) -> None:
def __init__(
self,
config: PermissionsConfig,
tools_config: ToolsConfig | None = None,
) -> None:
self.config = config
self._tools_config = tools_config
self._prompt_callback: PromptCallback | None = None
def set_prompt_callback(self, callback: PromptCallback) -> None:
@@ -36,9 +43,19 @@ class PermissionsService:
"""
self._prompt_callback = callback
async def check(self, tool_name: str, description: str = "") -> bool:
async def check(
self,
tool_name: str,
description: str = "",
arguments: str = "",
) -> bool:
"""Check if a tool is permitted to run.
Args:
tool_name: Name of the tool to check.
description: Human-readable description for the prompt.
arguments: Raw JSON arguments string (used for shell-aware checks).
Returns:
True if permitted, False if denied.
"""
@@ -50,6 +67,12 @@ class PermissionsService:
logger.debug("Tool '%s' is auto-approved", tool_name)
return True
# Shell-aware: check allowed/denied command lists
if tool_name == "run_command" and self._tools_config is not None:
result = self._check_shell_command(arguments)
if result is not None:
return result
# Prompt user via callback (TUI modal, etc.)
if self._prompt_callback is not None:
return await self._prompt_callback(tool_name, description)
@@ -57,3 +80,41 @@ class PermissionsService:
# No callback set — deny by default (safe fallback)
logger.warning("Tool '%s' requires approval but no prompt callback set — denied", tool_name)
return False
def _check_shell_command(self, arguments: str) -> bool | None:
"""Check shell command against allowed/denied lists.
Returns True (allow), False (deny), or None (fall through to prompt).
"""
shell_config = self._tools_config.shell # type: ignore[union-attr]
try:
cmd = json.loads(arguments).get("command", "")
except (json.JSONDecodeError, AttributeError):
return None # can't parse, fall through to prompt
try:
base_cmd = shlex.split(cmd)[0]
except (ValueError, IndexError):
return None
# Denied commands: prefix match on full command string
for denied in shell_config.denied_commands:
if cmd.startswith(denied):
logger.info("Shell command '%s' matches denied prefix '%s'", cmd, denied)
return False
# Allowed commands: base executable match
if shell_config.allowed_commands:
if base_cmd in shell_config.allowed_commands:
logger.debug(
"Shell command '%s' auto-approved (base '%s' in allowed list)",
cmd,
base_cmd,
)
return True
# Base command NOT in allowed list — fall through to prompt
return None
# No allowed list configured — fall through to prompt
return None

71
app/services/skills.py Normal file
View File

@@ -0,0 +1,71 @@
"""Skills manager — scans for and loads skill markdown files."""
from __future__ import annotations
import logging
from pathlib import Path
from pydantic import BaseModel
from app.models.config import SkillsConfig
logger = logging.getLogger(__name__)
class Skill(BaseModel):
"""Metadata for a discovered skill file."""
name: str
description: str
path: Path
class SkillsManager:
"""Discovers, indexes, and loads skill files from configured directories."""
def __init__(self, config: SkillsConfig, workspace_root: Path) -> None:
self._config = config
self._workspace = workspace_root
self._skills: dict[str, Skill] = {}
self._scan()
def _scan(self) -> None:
"""Scan configured directories for .md skill files."""
for skill_dir in self._config.directories:
resolved = (self._workspace / skill_dir) if not skill_dir.is_absolute() else skill_dir
if not resolved.is_dir():
logger.debug("Skills directory does not exist: %s", resolved)
continue
for md in sorted(resolved.glob("*.md")):
name = md.stem
desc = self._extract_description(md)
self._skills[name] = Skill(name=name, description=desc, path=md)
logger.debug("Discovered skill: %s (%s)", name, desc)
@staticmethod
def _extract_description(path: Path) -> str:
"""Extract the first non-blank, non-heading line as the description."""
for line in path.read_text().splitlines():
stripped = line.strip()
if stripped and not stripped.startswith("#"):
return stripped
return "(no description)"
def list_skills(self) -> list[Skill]:
"""Return all discovered skills."""
return list(self._skills.values())
def load_skill(self, name: str) -> str | None:
"""Load the full content of a skill by name. Returns None if not found."""
skill = self._skills.get(name)
return skill.path.read_text() if skill else None
def get_system_prompt_snippet(self) -> str:
"""Generate a snippet for the system prompt listing available skills."""
if not self._skills:
return ""
lines = ["\nAvailable skills (invoke with /skill-name):"]
for s in self._skills.values():
lines.append(f" - /{s.name}: {s.description}")
lines.append("To use a skill's full instructions, call the load_skill tool.")
return "\n".join(lines)

19
app/tools/registry.py
View File

@@ -1,12 +1,17 @@
"""Tool registration and schema export."""
from __future__ import annotations
import logging
from pathlib import Path
from typing import Any
from typing import TYPE_CHECKING, Any
from app.models.config import AppConfig
from app.tools.base import BaseTool
if TYPE_CHECKING:
from app.services.skills import SkillsManager
logger = logging.getLogger(__name__)
@@ -36,7 +41,11 @@ class ToolRegistry:
return [tool.get_openai_schema() for tool in self._tools.values()]
def create_default_registry(workspace_root: Path, config: AppConfig) -> ToolRegistry:
def create_default_registry(
workspace_root: Path,
config: AppConfig,
skills_manager: SkillsManager | None = None,
) -> ToolRegistry:
"""Create a ToolRegistry populated with all built-in tools."""
# Read tools
from app.tools.filesystem import ListDirTool, ReadFileTool
@@ -81,4 +90,10 @@ def create_default_registry(workspace_root: Path, config: AppConfig) -> ToolRegi
# Control flow
registry.register(FinishTool(workspace_root, config))
# Skills (conditional)
if skills_manager is not None:
from app.tools.skills import LoadSkillTool
registry.register(LoadSkillTool(workspace_root, config, skills_manager))
return registry

60
app/tools/skills.py Normal file
View File

@@ -0,0 +1,60 @@
"""Load skill tool — allows the LLM to load skill instructions on demand."""
from __future__ import annotations
from pathlib import Path
from typing import Any, ClassVar
from pydantic import BaseModel, Field
from app.models.config import AppConfig
from app.models.tool_call import ToolResult, ToolResultStatus
from app.services.skills import SkillsManager
from app.tools.base import BaseTool
class LoadSkillParams(BaseModel):
"""Parameters for the load_skill tool."""
name: str = Field(description="Name of the skill to load")
class LoadSkillTool(BaseTool):
"""Load a skill's full instructions by name.
Use when a skill is relevant to the current task.
"""
name: ClassVar[str] = "load_skill"
description: ClassVar[str] = (
"Load a skill's full instructions by name. "
"Use when a skill is relevant to the current task."
)
params_model: ClassVar[type[BaseModel]] = LoadSkillParams
def __init__(
self,
workspace_root: Path,
config: AppConfig,
skills_manager: SkillsManager,
) -> None:
super().__init__(workspace_root, config)
self._skills = skills_manager
def execute(self, *, tool_call_id: str, **kwargs: Any) -> ToolResult:
skill_name: str = kwargs["name"]
content = self._skills.load_skill(skill_name)
if content is None:
available = [s.name for s in self._skills.list_skills()]
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=f"Unknown skill '{skill_name}'. Available: {available}",
)
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.SUCCESS,
output=content,
)

137
app/ui/app.py
View File

@@ -10,7 +10,7 @@ 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 Header, RichLog
from app.agent.context import SessionContext
from app.agent.loop import AgentLoop
@@ -20,8 +20,14 @@ 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 StatusBar, StreamingStatic
from app.utils.display import DisplayAdapter, render_user_message
from app.ui.widgets import (
HistoryInput,
PermissionModal,
SessionResumeModal,
StatusBar,
StreamingStatic,
)
from app.utils.display import DisplayAdapter
from app.utils.logging import get_logger, setup_logging_for_tui
if TYPE_CHECKING:
@@ -49,6 +55,8 @@ class SneakyCodeApp(App):
self._client: LLMClient | None = None
self._tool_registry = None
self._permissions: PermissionsService | None = None
self._debug_logger = None
self._skills_manager = None
self._current_worker: Worker | None = None
self._cancel_count = 0
self.sub_title = config.llm.model
@@ -58,7 +66,7 @@ class SneakyCodeApp(App):
yield RichLog(id="chat-log", highlight=True, markup=True)
yield StreamingStatic("", id="streaming")
yield StatusBar()
yield Input(placeholder="Enter your prompt...")
yield HistoryInput(placeholder="Enter your prompt...")
async def on_mount(self) -> None:
"""Initialize agent components after the app is mounted."""
@@ -69,8 +77,28 @@ class SneakyCodeApp(App):
# Create long-lived agent dependencies (reused across turns)
self._client = LLMClient(self._config.llm)
await self._client.__aenter__()
self._tool_registry = create_default_registry(self._config.agent.workspace_root, self._config)
self._permissions = PermissionsService(self._config.permissions)
self._permissions = PermissionsService(self._config.permissions, self._config.tools)
# Create debug logger if enabled
if self._config.debug.enabled:
from app.services.debug_log import DebugLogger
log_dir = self._config.agent.workspace_root / self._config.debug.log_dir
self._debug_logger = DebugLogger(log_dir, self._config.debug.max_files)
# Initialize skills system
if self._config.skills.enabled:
from app.services.skills import SkillsManager
self._skills_manager = SkillsManager(
self._config.skills, self._config.agent.workspace_root
)
self._tool_registry = create_default_registry(
self._config.agent.workspace_root,
self._config,
skills_manager=self._skills_manager,
)
# Set up permission prompt callback
async def permission_prompt(tool_name: str, description: str) -> bool:
@@ -82,14 +110,16 @@ class SneakyCodeApp(App):
if self._session_mgr and self._config.session.offer_resume:
saved = self._session_mgr.load_latest()
if saved:
msg_count = len(saved.messages)
log = self.query_one("#chat-log", RichLog)
log.write(Text(f"Found previous session ({msg_count} messages)", style="cyan"))
# Auto-resume for now (modal can be added later)
self._session_mgr.restore(saved, self._ctx)
log.write(Text(f"✓ Session restored ({msg_count} messages)", style="bold green"))
msg_count = len(saved.messages)
resume = await self.push_screen_wait(SessionResumeModal(msg_count))
if resume:
self._session_mgr.restore(saved, self._ctx)
log.write(Text("Session restored", style="bold green"))
else:
log.write(Text("Starting fresh session", style="cyan"))
self.query_one(Input).focus()
self.query_one(HistoryInput).focus()
async def on_input_submitted(self, event: Input.Submitted) -> None:
"""Handle user input submission."""
@@ -98,6 +128,7 @@ class SneakyCodeApp(App):
return
event.input.clear()
event.input.record(user_input)
log = self.query_one("#chat-log", RichLog)
# Handle slash commands
@@ -105,9 +136,6 @@ class SneakyCodeApp(App):
await self._handle_slash_command(user_input, log)
return
# Write user message to log
log.write(render_user_message(user_input))
# Dispatch agent turn as async worker
self._cancel_count = 0
self._current_worker = self.run_worker(
@@ -145,7 +173,58 @@ class SneakyCodeApp(App):
f"Started: {self._ctx.start_time.isoformat()}",
style="cyan",
))
elif cmd.startswith("/models"):
parts = command.split(maxsplit=1)
if len(parts) == 1:
# List available models
try:
from app.services.llm import LLMError
models = await self._client.list_models()
from rich.table import Table
table = Table(title="Available Models", show_lines=False)
table.add_column("Model", style="cyan")
table.add_column("Size", style="dim")
current = self._config.llm.model
for m in models:
name = m["name"]
marker = " (active)" if current in name or name.startswith(current) else ""
table.add_row(f"{name}{marker}", m["size"])
log.write(table)
except Exception as e:
log.write(Text(f"Failed to list models: {e}", style="red"))
else:
new_model = parts[1].strip()
self._config.llm.model = new_model
self.sub_title = new_model
log.write(Text(f"Switched to model: {new_model}", style="bold green"))
elif cmd == "/skills":
if self._skills_manager:
skills = self._skills_manager.list_skills()
if not skills:
log.write(Text("No skills found", style="yellow"))
else:
from rich.table import Table
table = Table(title="Available Skills")
table.add_column("Name", style="cyan")
table.add_column("Description")
for s in skills:
table.add_row(f"/{s.name}", s.description)
log.write(table)
else:
log.write(Text("Skills system is disabled", style="yellow"))
else:
# Try as skill invocation
skill_name = cmd.lstrip("/")
if self._skills_manager:
content = self._skills_manager.load_skill(skill_name)
if content is not None:
if self._ctx:
self._ctx.add_message("system", f"[Skill: {skill_name}]\n{content}")
log.write(Text(f"Loaded skill: {skill_name}", style="bold green"))
return
log.write(Text(f"⚠ Unknown command: {command}", style="yellow"))
async def _run_agent_turn(self, user_input: str) -> None:
@@ -155,17 +234,21 @@ class SneakyCodeApp(App):
log = self.query_one("#chat-log", RichLog)
streaming_widget = self.query_one("#streaming", StreamingStatic)
status_bar = self.query_one(StatusBar)
display = DisplayAdapter(log)
# StreamHandler is per-turn (has per-turn accumulators)
handler = StreamHandler(self._config.display)
status_bar.start_streaming()
# Set up streaming UI callbacks
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)
def on_thinking() -> None:
streaming_widget.update(Text("Thinking...", style="dim"))
@@ -173,30 +256,28 @@ class SneakyCodeApp(App):
def on_done() -> None:
streaming_widget.hide_streaming()
status_bar.stop_streaming()
handler.set_callbacks(on_content=on_content, on_thinking=on_thinking, on_done=on_done)
agent = AgentLoop(
self._config, self._ctx, self._client, handler,
self._tool_registry, self._permissions, display,
debug_logger=self._debug_logger,
skills_manager=self._skills_manager,
)
await agent.run_turn(user_input)
status_bar.stop_streaming()
# Auto-save
if self._config.session.auto_save:
self._save_session()
async def _show_permission_modal(self, tool_name: str, description: str) -> bool:
"""Show a permission prompt in the chat log and wait for response.
Uses the Input widget to capture y/n response.
"""
# For v1, auto-approve all tools that reach the prompt
# TODO: Implement proper modal dialog
log = self.query_one("#chat-log", RichLog)
log.write(Text(f"⚠ Auto-approved: {tool_name} ({description})", style="yellow"))
return True
"""Show a modal dialog for tool permission approval."""
return await self.push_screen_wait(PermissionModal(tool_name, description))
def action_cancel_or_quit(self) -> None:
"""Handle Ctrl+C: first press cancels worker, second press quits."""
@@ -209,6 +290,11 @@ class SneakyCodeApp(App):
log = self.query_one("#chat-log", RichLog)
log.write(Text("⚠ Cancelling... (press Ctrl+C again to quit)", style="yellow"))
async def on_unmount(self) -> None:
"""Clean up the LLM client on app shutdown."""
if self._client is not None:
await self._client.close()
def on_worker_state_changed(self, event: Worker.StateChanged) -> None:
"""Handle worker completion or failure."""
from textual.worker import WorkerState
@@ -223,9 +309,10 @@ class SneakyCodeApp(App):
if event.state in (WorkerState.SUCCESS, WorkerState.ERROR, WorkerState.CANCELLED):
self._current_worker = None
# Hide streaming widget in case it was left visible
# Hide streaming widget and stop spinner in case they were left active
streaming = self.query_one("#streaming", StreamingStatic)
streaming.hide_streaming()
self.query_one(StatusBar).stop_streaming()
def _save_session(self) -> Path | None:
"""Save session quietly, return path or None."""

29
app/ui/styles.tcss
View File

@@ -21,6 +21,35 @@ Screen {
display: block;
}
/* Modal dialog */
#permission-dialog {
width: 60;
height: auto;
max-height: 12;
border: thick $accent;
background: $surface;
padding: 1 2;
}
.modal-title {
text-style: bold;
margin-bottom: 1;
}
.modal-body {
margin-bottom: 1;
color: $text-muted;
}
.modal-buttons {
height: 3;
align: center middle;
}
.modal-buttons Button {
margin: 0 1;
}
/* StatusBar styles are in DEFAULT_CSS on the widget itself */
Input {

162
app/ui/widgets.py
View File

@@ -2,12 +2,130 @@
from __future__ import annotations
from textual.app import ComposeResult
from textual.containers import Horizontal, Vertical
from textual.events import Key
from textual.screen import ModalScreen
from textual.timer import Timer
from textual.widgets import Button, Input, Static
from rich.text import Text
from textual.widgets import Static
# ---------------------------------------------------------------------------
# Modal Dialogs
# ---------------------------------------------------------------------------
class PermissionModal(ModalScreen[bool]):
"""Modal dialog for tool permission approval."""
BINDINGS = [("y", "allow", "Allow"), ("n", "deny", "Deny")]
def __init__(self, tool_name: str, description: str) -> None:
super().__init__()
self._tool_name = tool_name
self._description = description
def compose(self) -> ComposeResult:
with Vertical(id="permission-dialog"):
yield Static(f"Tool: {self._tool_name}", classes="modal-title")
yield Static(self._description, classes="modal-body")
with Horizontal(classes="modal-buttons"):
yield Button("Allow (y)", id="allow", variant="success")
yield Button("Deny (n)", id="deny", variant="error")
def on_button_pressed(self, event: Button.Pressed) -> None:
self.dismiss(event.button.id == "allow")
def action_allow(self) -> None:
self.dismiss(True)
def action_deny(self) -> None:
self.dismiss(False)
class SessionResumeModal(ModalScreen[bool]):
"""Modal dialog for session resume prompt."""
BINDINGS = [("y", "resume", "Resume"), ("n", "fresh", "Start Fresh")]
def __init__(self, msg_count: int) -> None:
super().__init__()
self._msg_count = msg_count
def compose(self) -> ComposeResult:
with Vertical(id="permission-dialog"):
yield Static("Resume Session?", classes="modal-title")
yield Static(
f"Found previous session with {self._msg_count} messages.",
classes="modal-body",
)
with Horizontal(classes="modal-buttons"):
yield Button("Resume (y)", id="resume", variant="success")
yield Button("Start Fresh (n)", id="fresh", variant="warning")
def on_button_pressed(self, event: Button.Pressed) -> None:
self.dismiss(event.button.id == "resume")
def action_resume(self) -> None:
self.dismiss(True)
def action_fresh(self) -> None:
self.dismiss(False)
# ---------------------------------------------------------------------------
# Input with History
# ---------------------------------------------------------------------------
class HistoryInput(Input):
"""Input with up/down arrow history cycling."""
def __init__(self, **kwargs: object) -> None:
super().__init__(**kwargs)
self._history: list[str] = []
self._history_index: int = -1
self._draft: str = ""
def record(self, value: str) -> None:
"""Record a submitted value in history."""
if value.strip() and (not self._history or self._history[-1] != value):
self._history.append(value)
self._history_index = -1
self._draft = ""
def on_key(self, event: Key) -> None:
if event.key == "up" and self._history:
event.prevent_default()
if self._history_index == -1:
self._draft = self.value
self._history_index = len(self._history) - 1
elif self._history_index > 0:
self._history_index -= 1
self.value = self._history[self._history_index]
self.cursor_position = len(self.value)
elif event.key == "down" and self._history_index != -1:
event.prevent_default()
self._history_index += 1
if self._history_index >= len(self._history):
self._history_index = -1
self.value = self._draft
else:
self.value = self._history[self._history_index]
self.cursor_position = len(self.value)
# ---------------------------------------------------------------------------
# Status Bar
# ---------------------------------------------------------------------------
class StatusBar(Static):
"""Single-line status bar showing token usage and iteration count."""
"""Single-line status bar showing token usage, iteration count, and streaming spinner."""
_SPINNER = "\u280b\u2819\u2839\u2838\u283c\u2834\u2826\u2827\u2807\u280f"
DEFAULT_CSS = """
StatusBar {
@@ -25,6 +143,10 @@ class StatusBar(Static):
self._budget: int = 0
self._iteration: int = 0
self._max_iterations: int = 0
self._streaming: bool = False
self._spinner_frame: int = 0
self._spinner_timer: Timer | None = None
self._stream_tokens: int = 0
def update_tokens(self, tokens: int, budget: int) -> None:
"""Update the token usage display."""
@@ -38,14 +160,48 @@ class StatusBar(Static):
self._max_iterations = max_iterations
self._refresh_display()
def start_streaming(self) -> None:
"""Start the animated thinking spinner."""
self._streaming = True
self._spinner_frame = 0
self._stream_tokens = 0
self._spinner_timer = self.set_interval(0.08, self._tick_spinner)
self._refresh_display()
def stop_streaming(self) -> None:
"""Stop the animated thinking spinner."""
self._streaming = False
if self._spinner_timer:
self._spinner_timer.stop()
self._spinner_timer = None
self._refresh_display()
def update_stream_tokens(self, tokens: int) -> None:
"""Update estimated token count during streaming."""
self._stream_tokens = tokens
def _tick_spinner(self) -> None:
self._spinner_frame = (self._spinner_frame + 1) % len(self._SPINNER)
self._refresh_display()
def _refresh_display(self) -> None:
"""Rebuild the status bar text."""
parts: list[str] = []
if self._streaming:
spinner = self._SPINNER[self._spinner_frame]
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(" │ ".join(parts), style="dim"))
self.update(Text(" \u2502 ".join(parts), style="dim"))
# ---------------------------------------------------------------------------
# Streaming Display
# ---------------------------------------------------------------------------
class StreamingStatic(Static):

10
config/config.yaml
View File

@@ -71,3 +71,13 @@ display:
show_tool_calls: true
show_token_usage: true
stream_output: true
debug:
enabled: false
log_dir: ".sneakycode/logs"
max_files: 10
skills:
enabled: true
directories:
- ".sneakycode/skills"