feat: implement tweaks plan - modals, smart shell, spinner, /models, debug log, skills

Phase 1: Permission modal dialog, session resume modal, HistoryInput with
up/down arrow cycling, remove "You:" echo from chat log, LLM client cleanup
on unmount.

Phase 2: Smart shell auto-approve using allowed/denied command lists from
ToolsConfig, animated thinking spinner with live token count in status bar.

Phase 3: /models slash command (list + switch), CLI directory positional
argument, JSONL debug logger with rotation.

Phase 4: Skills system with SkillsManager, load_skill LLM tool, /skills
listing, skill invocation via slash commands, system prompt integration.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-11 15:46:44 -05:00
parent 7600195ecf
commit 3f9012e6c2
13 changed files with 683 additions and 37 deletions

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()

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.

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)