feat: structured skill packages with config overrides, chaining, and TUI integration

Add a skill package system where each skill is a directory with a skill.yaml
manifest and prompt markdown files. Skills support /command triggers, scoped
config overrides (temperature, max_tokens, tool filtering), chain dependencies
with cycle-safe resolution, and a finish_skill completion signal.

Includes four built-in skills: explore, brainstorm, write-document, and plan.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-11 19:06:05 -05:00
parent 26bcbc6c1f
commit 2ae8294e29
16 changed files with 832 additions and 31 deletions

View File

@@ -19,6 +19,7 @@ from app.utils.logging import get_logger
if TYPE_CHECKING:
from app.services.debug_log import DebugLogger
from app.services.skill_runner import SkillRunner
from app.services.skills import SkillsManager
logger = get_logger(__name__)
@@ -45,6 +46,7 @@ class AgentLoop:
display: DisplayAdapter | None = None,
debug_logger: DebugLogger | None = None,
skills_manager: SkillsManager | None = None,
skill_runner: SkillRunner | None = None,
) -> None:
self._config = config
self._ctx = ctx
@@ -55,6 +57,7 @@ class AgentLoop:
self._display = display
self._debug = debug_logger
self._skills = skills_manager
self._skill_runner = skill_runner
self._tools_schema = registry.get_openai_tools_schema()
self._system_prompt = self._build_system_prompt()
self._cancelled = False
@@ -81,6 +84,11 @@ class AgentLoop:
)
if self._skills:
prompt += self._skills.get_system_prompt_snippet()
if self._skill_runner and self._skill_runner.is_active:
prompt += (
f"\n\nCurrently active skill: {self._skill_runner.active_skill_name}. "
"When the skill's objective is complete, call the `finish_skill` tool."
)
return prompt
def _get_messages_with_system_prompt(self) -> list[Message]:
@@ -199,7 +207,12 @@ class AgentLoop:
name=result.tool_name,
)
# Check if finish tool was called
# Rebuild tools schema and system prompt if skill state may have changed
if any(r.tool_name in ("load_skill", "finish_skill") for r in results):
self._tools_schema = self._registry.get_openai_tools_schema()
self._system_prompt = self._build_system_prompt()
# Check if finish tool was called (finish_skill does NOT break the loop)
if any(r.tool_name == "finish" for r in results):
break
else:

39
app/models/skill.py Normal file
View File

@@ -0,0 +1,39 @@
"""Pydantic models for structured skill packages."""
from __future__ import annotations
from pydantic import BaseModel, Field
class SkillConfigOverrides(BaseModel):
"""Scoped config overrides applied while a skill is active."""
temperature: float | None = Field(default=None, description="Override sampling temperature")
max_tokens: int | None = Field(default=None, description="Override max tokens")
tools_enable: list[str] | None = Field(
default=None, description="Whitelist — only these tools available when set"
)
tools_disable: list[str] | None = Field(
default=None, description="Blacklist — disable specific tools"
)
class SkillManifest(BaseModel):
"""Parsed skill.yaml manifest for a skill package directory."""
name: str = Field(description="Unique skill identifier")
description: str = Field(description="Human-readable skill description")
version: str = Field(default="1.0", description="Skill version")
triggers: list[str] = Field(
default_factory=list, description="Slash commands that activate this skill"
)
config_overrides: SkillConfigOverrides = Field(
default_factory=SkillConfigOverrides, description="Scoped config overrides"
)
chain: list[str] = Field(
default_factory=list, description="Skill names to run first (dependencies)"
)
prompts: list[str] = Field(
default_factory=lambda: ["prompt.md"],
description="Markdown prompt files to load, in order",
)

View File

@@ -0,0 +1,234 @@
"""SkillRunner — orchestrates skill activation, chaining, config scoping, and deactivation."""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from app.agent.context import SessionContext
from app.models.config import AppConfig
from app.models.skill import SkillManifest
from app.services.skills import Skill, SkillsManager
from app.tools.registry import ToolRegistry
logger = logging.getLogger(__name__)
class SkillChainError(Exception):
"""Raised when skill chain resolution fails (e.g., cycle detected)."""
@dataclass
class _SkillSnapshot:
"""Captured state before skill activation, for restoration on deactivate."""
temperature: float
max_tokens: int
disabled_tools: set[str] = field(default_factory=set)
class SkillRunner:
"""Manages skill lifecycle: activation, chaining, config overrides, deactivation.
Only one skill can be active at a time. Activating a new skill while one
is active will first deactivate the current skill.
"""
def __init__(
self,
skills_manager: SkillsManager,
config: AppConfig,
ctx: SessionContext,
registry: ToolRegistry,
) -> None:
self._skills = skills_manager
self._config = config
self._ctx = ctx
self._registry = registry
self._active_skill: Skill | None = None
self._snapshot: _SkillSnapshot | None = None
@property
def is_active(self) -> bool:
"""Whether a skill is currently active."""
return self._active_skill is not None
@property
def active_skill_name(self) -> str | None:
"""Name of the currently active skill, or None."""
return self._active_skill.name if self._active_skill else None
@property
def active_skill(self) -> Skill | None:
"""The currently active skill, or None."""
return self._active_skill
def activate(self, skill_name: str) -> str | None:
"""Activate a skill by name.
Resolves chain dependencies (depth-first), applies config overrides,
injects prompt content into conversation context.
Args:
skill_name: Name of the skill to activate.
Returns:
The concatenated prompt content injected, or None on failure.
Raises:
SkillChainError: If chain resolution detects a cycle.
"""
skill = self._skills.get_skill(skill_name)
if skill is None:
logger.warning("Cannot activate unknown skill: %s", skill_name)
return None
# Deactivate current skill if one is active
if self._active_skill is not None:
self.deactivate()
# Resolve chain dependencies
chain = self._resolve_chain(skill, set())
# Snapshot current config for restoration
self._snapshot = _SkillSnapshot(
temperature=self._config.llm.temperature,
max_tokens=self._config.llm.max_tokens,
)
# Collect and inject chain skill prompts first
all_prompts: list[str] = []
for chained_skill in chain:
content = self._skills.load_skill(chained_skill.name)
if content:
all_prompts.append(f"[Chained skill: {chained_skill.name}]\n{content}")
# Load the target skill's prompts
content = self._skills.load_skill(skill.name)
if content:
all_prompts.append(content)
# Apply config overrides from the target skill
if skill.manifest:
self._apply_overrides(skill.manifest)
# Inject prompts into context
full_prompt = "\n\n".join(all_prompts) if all_prompts else None
if full_prompt:
self._ctx.add_message(
"system",
f"[Skill activated: {skill.name}]\n{full_prompt}",
)
self._active_skill = skill
logger.info("Skill activated: %s", skill.name)
return full_prompt
def activate_by_trigger(self, trigger: str) -> str | None:
"""Activate a skill by its /command trigger.
Args:
trigger: The trigger string (with or without leading /).
Returns:
The concatenated prompt content, or None if no skill matches.
"""
skill = self._skills.get_skill_by_trigger(trigger)
if skill is None:
return None
return self.activate(skill.name)
def deactivate(self, summary: str | None = None) -> None:
"""Deactivate the current skill, restoring config and tool state.
Args:
summary: Optional summary message to inject into context.
"""
if self._active_skill is None:
return
skill_name = self._active_skill.name
# Restore config
if self._snapshot is not None:
self._config.llm.temperature = self._snapshot.temperature
self._config.llm.max_tokens = self._snapshot.max_tokens
self._registry.restore_filter(self._snapshot.disabled_tools)
self._snapshot = None
if summary:
self._ctx.add_message(
"system",
f"[Skill completed: {skill_name}] {summary}",
)
self._active_skill = None
logger.info("Skill deactivated: %s", skill_name)
def _resolve_chain(
self, skill: Skill, in_progress: set[str], completed: set[str] | None = None,
) -> list[Skill]:
"""Depth-first resolution of skill chain dependencies.
Uses separate in_progress (current path) and completed sets to correctly
handle diamond dependencies without false cycle detection.
Args:
skill: The skill whose chain to resolve.
in_progress: Skills on the current recursion path (for cycle detection).
completed: Skills already fully resolved (skip duplicates).
Returns:
Ordered list of chained skills to activate before the target.
Raises:
SkillChainError: If a cycle is detected.
"""
if completed is None:
completed = set()
if skill.manifest is None or not skill.manifest.chain:
return []
result: list[Skill] = []
for dep_name in skill.manifest.chain:
if dep_name in completed:
continue # Already resolved via another branch (diamond dep)
if dep_name in in_progress:
raise SkillChainError(
f"Cycle detected in skill chain: {dep_name} already in progress "
f"(path: {' -> '.join(in_progress)} -> {dep_name})"
)
dep_skill = self._skills.get_skill(dep_name)
if dep_skill is None:
logger.warning("Chained skill not found: %s (required by %s)", dep_name, skill.name)
continue
in_progress.add(dep_name)
result.extend(self._resolve_chain(dep_skill, in_progress, completed))
in_progress.discard(dep_name)
completed.add(dep_name)
result.append(dep_skill)
return result
def _apply_overrides(self, manifest: SkillManifest) -> None:
"""Apply config overrides from a skill manifest."""
overrides = manifest.config_overrides
if overrides.temperature is not None:
self._config.llm.temperature = overrides.temperature
if overrides.max_tokens is not None:
self._config.llm.max_tokens = overrides.max_tokens
if overrides.tools_enable is not None or overrides.tools_disable is not None:
previous = self._registry.apply_filter(
enable=overrides.tools_enable,
disable=overrides.tools_disable,
)
# Store for restoration
if self._snapshot:
self._snapshot.disabled_tools = previous

View File

@@ -1,46 +1,92 @@
"""Skills manager — scans for and loads skill markdown files."""
"""Skills manager — scans for and loads skill packages and legacy markdown files."""
from __future__ import annotations
import logging
from pathlib import Path
from pydantic import BaseModel
import yaml
from pydantic import BaseModel, ValidationError
from app.models.config import SkillsConfig
from app.models.skill import SkillManifest
logger = logging.getLogger(__name__)
class Skill(BaseModel):
"""Metadata for a discovered skill file."""
"""Metadata for a discovered skill (package or legacy flat file)."""
name: str
description: str
path: Path
manifest: SkillManifest | None = None
class SkillsManager:
"""Discovers, indexes, and loads skill files from configured directories."""
"""Discovers, indexes, and loads skill files from configured directories.
Supports both:
- Directory-based packages (contain skill.yaml + prompt .md files)
- Legacy flat .md files (backwards compatible)
"""
def __init__(self, config: SkillsConfig, workspace_root: Path) -> None:
self._config = config
self._workspace = workspace_root
self._skills: dict[str, Skill] = {}
self._trigger_map: dict[str, str] = {} # trigger -> skill name
self._scan()
def _scan(self) -> None:
"""Scan configured directories for .md skill files."""
"""Scan configured directories for skill packages and legacy .md 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)
for entry in sorted(resolved.iterdir()):
if entry.is_dir():
self._scan_package(entry)
elif entry.suffix == ".md":
self._scan_legacy(entry)
def _scan_package(self, pkg_dir: Path) -> None:
"""Scan a directory-based skill package containing skill.yaml."""
manifest_path = pkg_dir / "skill.yaml"
if not manifest_path.exists():
logger.debug("Skipping directory without skill.yaml: %s", pkg_dir)
return
try:
raw = yaml.safe_load(manifest_path.read_text())
manifest = SkillManifest(**raw)
except (yaml.YAMLError, ValidationError, TypeError) as e:
logger.warning("Failed to parse skill manifest %s: %s", manifest_path, e)
return
skill = Skill(
name=manifest.name,
description=manifest.description,
path=pkg_dir,
manifest=manifest,
)
self._skills[manifest.name] = skill
# Register triggers
for trigger in manifest.triggers:
normalized = trigger.lstrip("/").lower()
self._trigger_map[normalized] = manifest.name
logger.debug("Discovered skill package: %s (%s)", manifest.name, manifest.description)
def _scan_legacy(self, md_path: Path) -> None:
"""Scan a legacy flat .md skill file."""
name = md_path.stem
desc = self._extract_description(md_path)
self._skills[name] = Skill(name=name, description=desc, path=md_path)
logger.debug("Discovered legacy skill: %s (%s)", name, desc)
@staticmethod
def _extract_description(path: Path) -> str:
@@ -55,10 +101,51 @@ class SkillsManager:
"""Return all discovered skills."""
return list(self._skills.values())
def get_skill(self, name: str) -> Skill | None:
"""Look up a skill by name."""
return self._skills.get(name)
def get_skill_by_trigger(self, trigger: str) -> Skill | None:
"""Look up a skill by /command trigger.
Args:
trigger: The trigger string (with or without leading /).
Returns:
The matching Skill, or None.
"""
normalized = trigger.lstrip("/").lower()
skill_name = self._trigger_map.get(normalized)
if skill_name:
return self._skills.get(skill_name)
return None
def load_skill(self, name: str) -> str | None:
"""Load the full content of a skill by name. Returns None if not found."""
"""Load the full content of a skill by name.
For package skills, concatenates all prompt .md files.
For legacy skills, returns the .md file content.
Returns:
Concatenated prompt content, or None if not found.
"""
skill = self._skills.get(name)
return skill.path.read_text() if skill else None
if skill is None:
return None
if skill.manifest is not None:
# Package skill: load prompt files
parts: list[str] = []
for prompt_file in skill.manifest.prompts:
prompt_path = skill.path / prompt_file
if prompt_path.exists():
parts.append(prompt_path.read_text())
else:
logger.warning("Prompt file not found: %s", prompt_path)
return "\n\n".join(parts) if parts else None
else:
# Legacy flat file
return skill.path.read_text()
def get_system_prompt_snippet(self) -> str:
"""Generate a snippet for the system prompt listing available skills."""
@@ -66,6 +153,10 @@ class SkillsManager:
return ""
lines = ["\nAvailable skills (invoke with /skill-name):"]
for s in self._skills.values():
lines.append(f" - /{s.name}: {s.description}")
if s.manifest and s.manifest.triggers:
trigger_str = ", ".join(s.manifest.triggers)
lines.append(f" - {trigger_str}: {s.description}")
else:
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)

View File

@@ -20,6 +20,7 @@ class ToolRegistry:
def __init__(self) -> None:
self._tools: dict[str, BaseTool] = {}
self._disabled: set[str] = set()
def register(self, tool: BaseTool) -> None:
"""Register a tool instance. Raises ValueError on duplicate name."""
@@ -29,24 +30,74 @@ class ToolRegistry:
logger.debug("Registered tool: %s", tool.name)
def get(self, name: str) -> BaseTool | None:
"""Look up a tool by name."""
"""Look up a tool by name. Returns None if disabled or not found."""
if name in self._disabled:
return None
return self._tools.get(name)
def get_all(self) -> dict[str, BaseTool]:
"""Return all registered tools."""
return dict(self._tools)
"""Return all registered tools (excluding disabled)."""
return {k: v for k, v in self._tools.items() if k not in self._disabled}
def get_openai_tools_schema(self) -> list[dict[str, Any]]:
"""Return OpenAI function-calling schemas for all registered tools."""
return [tool.get_openai_schema() for tool in self._tools.values()]
"""Return OpenAI function-calling schemas for all active tools."""
return [
tool.get_openai_schema()
for tool in self._tools.values()
if tool.name not in self._disabled
]
def apply_filter(
self,
*,
enable: list[str] | None = None,
disable: list[str] | None = None,
) -> set[str]:
"""Apply a tool filter, returning the previous disabled set for restoration.
Args:
enable: If set, only these tools (plus always-on tools) are available.
disable: Specific tools to disable.
Returns:
The previous disabled set (snapshot for restore).
"""
previous = set(self._disabled)
if enable is not None:
# Whitelist mode: disable everything not in the enable list
self._disabled = {name for name in self._tools if name not in enable}
elif disable is not None:
# Blacklist mode: add to existing disabled set (preserves global disables)
self._disabled = set(self._disabled) | set(disable)
else:
self._disabled = set()
return previous
def restore_filter(self, previous: set[str]) -> None:
"""Restore a previous filter state."""
self._disabled = previous
def all_tool_names(self) -> list[str]:
"""Return all registered tool names (including disabled)."""
return list(self._tools.keys())
def create_default_registry(
workspace_root: Path,
config: AppConfig,
skills_manager: SkillsManager | None = None,
skill_runner: object | None = None,
) -> ToolRegistry:
"""Create a ToolRegistry populated with all built-in tools."""
"""Create a ToolRegistry populated with all built-in tools.
Args:
workspace_root: Workspace root path.
config: Application configuration.
skills_manager: Optional skills manager for skill tools.
skill_runner: Optional SkillRunner for package skill activation.
"""
# Read tools
from app.tools.filesystem import ListDirTool, ReadFileTool
@@ -92,8 +143,11 @@ def create_default_registry(
# Skills (conditional)
if skills_manager is not None:
from app.tools.skills import LoadSkillTool
from app.services.skill_runner import SkillRunner as SkillRunnerType
from app.tools.skills import FinishSkillTool, LoadSkillTool
registry.register(LoadSkillTool(workspace_root, config, skills_manager))
runner = skill_runner if isinstance(skill_runner, SkillRunnerType) else None
registry.register(LoadSkillTool(workspace_root, config, skills_manager, runner))
registry.register(FinishSkillTool(workspace_root, config, runner))
return registry

View File

@@ -1,17 +1,20 @@
"""Load skill tool — allows the LLM to load skill instructions on demand."""
"""Skill toolsload and finish skills during agent operation."""
from __future__ import annotations
from pathlib import Path
from typing import Any, ClassVar
from typing import TYPE_CHECKING, 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
if TYPE_CHECKING:
from app.services.skill_runner import SkillRunner
from app.services.skills import SkillsManager
class LoadSkillParams(BaseModel):
"""Parameters for the load_skill tool."""
@@ -23,6 +26,8 @@ class LoadSkillTool(BaseTool):
"""Load a skill's full instructions by name.
Use when a skill is relevant to the current task.
For package skills, this activates the full skill lifecycle
(config overrides, chaining, prompt injection).
"""
name: ClassVar[str] = "load_skill"
@@ -37,14 +42,22 @@ class LoadSkillTool(BaseTool):
workspace_root: Path,
config: AppConfig,
skills_manager: SkillsManager,
skill_runner: SkillRunner | None = None,
) -> None:
super().__init__(workspace_root, config)
self._skills = skills_manager
self._runner = skill_runner
def set_skill_runner(self, runner: SkillRunner) -> None:
"""Late-bind the SkillRunner (avoids circular init dependencies)."""
self._runner = runner
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:
# Check if skill exists
skill = self._skills.get_skill(skill_name)
if skill is None:
available = [s.name for s in self._skills.list_skills()]
return ToolResult(
tool_call_id=tool_call_id,
@@ -52,9 +65,94 @@ class LoadSkillTool(BaseTool):
status=ToolResultStatus.ERROR,
error=f"Unknown skill '{skill_name}'. Available: {available}",
)
# For package skills with a runner, use full activation flow
if skill.manifest is not None and self._runner is not None:
content = self._runner.activate(skill_name)
if content is None:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=f"Failed to activate skill '{skill_name}'",
)
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.SUCCESS,
output=f"Skill '{skill_name}' activated.\n\n{content}",
)
# Legacy skill: just load content
content = self._skills.load_skill(skill_name)
if content is None:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=f"Failed to load skill '{skill_name}'",
)
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.SUCCESS,
output=content,
)
class FinishSkillParams(BaseModel):
"""Parameters for the finish_skill tool."""
summary: str = Field(
default="Skill complete.",
description="Brief summary of what was accomplished during the skill",
)
class FinishSkillTool(BaseTool):
"""Signal that the active skill is complete and should be deactivated.
Restores config overrides and tool availability to pre-skill state.
The agent loop continues after this (unlike the finish tool).
"""
name: ClassVar[str] = "finish_skill"
description: ClassVar[str] = (
"Call this when the active skill's task is complete. "
"Deactivates the skill and restores normal config. "
"The conversation continues after this."
)
params_model: ClassVar[type[BaseModel]] = FinishSkillParams
def __init__(
self,
workspace_root: Path,
config: AppConfig,
skill_runner: SkillRunner | None = None,
) -> None:
super().__init__(workspace_root, config)
self._runner = skill_runner
def set_skill_runner(self, runner: SkillRunner) -> None:
"""Late-bind the SkillRunner (avoids circular init dependencies)."""
self._runner = runner
def execute(self, *, tool_call_id: str, **kwargs: Any) -> ToolResult:
summary: str = kwargs.get("summary", "Skill complete.")
if self._runner is None or not self._runner.is_active:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error="No skill is currently active.",
)
skill_name = self._runner.active_skill_name
self._runner.deactivate(summary=summary)
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.SUCCESS,
output=f"Skill '{skill_name}' completed: {summary}",
)

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, RichLog
from textual.widgets import Header, Input, RichLog
from textual import work
from app.agent.context import SessionContext
@@ -58,6 +58,7 @@ class SneakyCodeApp(App):
self._permissions: PermissionsService | None = None
self._debug_logger = None
self._skills_manager = None
self._skill_runner = None
self._current_worker: Worker | None = None
self._cancel_count = 0
self.sub_title = config.llm.model
@@ -95,12 +96,31 @@ class SneakyCodeApp(App):
self._config.skills, self._config.agent.workspace_root
)
# Create tool registry (SkillRunner wired after registry exists)
self._tool_registry = create_default_registry(
self._config.agent.workspace_root,
self._config,
skills_manager=self._skills_manager,
)
# Create SkillRunner and late-bind it to skill tools
if self._skills_manager is not None and self._tool_registry is not None:
from app.services.skill_runner import SkillRunner
self._skill_runner = SkillRunner(
self._skills_manager,
self._config,
self._ctx,
self._tool_registry,
)
# Late-bind runner to skill tools already in the registry
load_tool = self._tool_registry.get("load_skill")
if load_tool and hasattr(load_tool, "set_skill_runner"):
load_tool.set_skill_runner(self._skill_runner)
finish_tool = self._tool_registry.get("finish_skill")
if finish_tool and hasattr(finish_tool, "set_skill_runner"):
finish_tool.set_skill_runner(self._skill_runner)
# Set up permission prompt callback
async def permission_prompt(tool_name: str, description: str) -> bool:
return await self._show_permission_modal(tool_name, description)
@@ -152,7 +172,24 @@ class SneakyCodeApp(App):
async def _handle_slash_command(self, command: str, log: RichLog) -> None:
"""Process slash commands."""
cmd = command.lower()
if cmd == "/quit":
if cmd == "/help":
from rich.table import Table
table = Table(title="SneakyCode Commands", show_lines=False)
table.add_column("Command", style="cyan", no_wrap=True)
table.add_column("Description")
table.add_row("/help", "Show this help message")
table.add_row("/quit, /exit, /bye", "Save session and exit")
table.add_row("/clear", "Clear conversation history")
table.add_row("/history", "Show conversation history")
table.add_row("/save", "Manually save session")
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("/skills", "List available skills")
table.add_row("/<skill>", "Load a skill by name")
log.write(table)
elif cmd in ("/quit", "/exit", "/bye"):
self._save_session()
self.exit()
elif cmd == "/clear":
@@ -221,7 +258,24 @@ class SneakyCodeApp(App):
else:
log.write(Text("Skills system is disabled", style="yellow"))
else:
# Try as skill invocation
# Try as skill trigger (package skill via SkillRunner)
if self._skill_runner and self._skills_manager:
skill = self._skills_manager.get_skill_by_trigger(cmd.lstrip("/"))
if skill is not None:
content = self._skill_runner.activate(skill.name)
status_bar = self.query_one(StatusBar)
status_bar.set_active_skill(skill.name)
log.write(Text(f"Skill activated: {skill.name}", style="bold green"))
# Run an agent turn so the LLM sees the skill context
self._cancel_count = 0
self._current_worker = self.run_worker(
self._run_agent_turn(f"[Skill activated: {skill.name}]"),
name="agent-turn",
exclusive=True,
)
return
# Try as legacy skill invocation
skill_name = cmd.lstrip("/")
if self._skills_manager:
content = self._skills_manager.load_skill(skill_name)
@@ -230,7 +284,7 @@ class SneakyCodeApp(App):
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"))
log.write(Text(f"Unknown command: {command}", style="yellow"))
async def _run_agent_turn(self, user_input: str) -> None:
"""Run a single agent turn (called as a worker)."""
@@ -270,12 +324,19 @@ class SneakyCodeApp(App):
self._tool_registry, self._permissions, display,
debug_logger=self._debug_logger,
skills_manager=self._skills_manager,
skill_runner=self._skill_runner,
)
await agent.run_turn(user_input)
status_bar.stop_streaming()
# 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)
elif self._skill_runner and self._skill_runner.is_active:
status_bar.set_active_skill(self._skill_runner.active_skill_name)
# Auto-save
if self._config.session.auto_save:
self._save_session()

View File

@@ -147,6 +147,7 @@ class StatusBar(Static):
self._spinner_frame: int = 0
self._spinner_timer: Timer | None = None
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."""
@@ -184,9 +185,16 @@ class StatusBar(Static):
self._spinner_frame = (self._spinner_frame + 1) % len(self._SPINNER)
self._refresh_display()
def set_active_skill(self, skill_name: str | None) -> None:
"""Set or clear the active skill indicator."""
self._active_skill = skill_name
self._refresh_display()
def _refresh_display(self) -> None:
"""Rebuild the status bar text."""
parts: list[str] = []
if self._active_skill:
parts.append(f"[Skill: {self._active_skill}]")
if self._streaming:
spinner = self._SPINNER[self._spinner_frame]
parts.append(f"{spinner} Thinking")