From ac3855e3947d6d4a47de984d5acabdad13c49312 Mon Sep 17 00:00:00 2001 From: Alan Chen <128259419+Chen-zexi@users.noreply.github.com> Date: Fri, 27 Mar 2026 10:44:48 -0400 Subject: [PATCH] feat(command): add `/skill` slash command for user-activated skill injection (#2488) * feat(command): add /skill slash command for user-activated skill injection * test(command): add tests for /skill slash command * refactor(command): switch skill activation from /skill prefix to $-reference interceptor --- nanobot/command/builtin.py | 82 ++++++++++- tests/cli/test_skill_command.py | 236 ++++++++++++++++++++++++++++++++ 2 files changed, 314 insertions(+), 4 deletions(-) create mode 100644 tests/cli/test_skill_command.py diff --git a/nanobot/command/builtin.py b/nanobot/command/builtin.py index 643397057..26c7273c1 100644 --- a/nanobot/command/builtin.py +++ b/nanobot/command/builtin.py @@ -4,6 +4,7 @@ from __future__ import annotations import asyncio import os +import re import sys from nanobot import __version__ @@ -11,6 +12,9 @@ from nanobot.bus.events import OutboundMessage from nanobot.command.router import CommandContext, CommandRouter from nanobot.utils.helpers import build_status_content +# Pattern to match $skill-name tokens (word chars + hyphens) +_SKILL_REF = re.compile(r"\$([A-Za-z][A-Za-z0-9_-]*)") + async def cmd_stop(ctx: CommandContext) -> OutboundMessage: """Cancel all active tasks and subagents for the session.""" @@ -56,8 +60,10 @@ async def cmd_status(ctx: CommandContext) -> OutboundMessage: channel=ctx.msg.channel, chat_id=ctx.msg.chat_id, content=build_status_content( - version=__version__, model=loop.model, - start_time=loop._start_time, last_usage=loop._last_usage, + version=__version__, + model=loop.model, + start_time=loop._start_time, + last_usage=loop._last_usage, context_window_tokens=loop.context_window_tokens, session_msg_count=len(session.get_history(max_messages=0)), context_tokens_estimate=ctx_est, @@ -70,18 +76,82 @@ async def cmd_new(ctx: CommandContext) -> OutboundMessage: """Start a fresh session.""" loop = ctx.loop session = ctx.session or loop.sessions.get_or_create(ctx.key) - snapshot = session.messages[session.last_consolidated:] + snapshot = session.messages[session.last_consolidated :] session.clear() loop.sessions.save(session) loop.sessions.invalidate(session.key) if snapshot: loop._schedule_background(loop.memory_consolidator.archive_messages(snapshot)) return OutboundMessage( - channel=ctx.msg.channel, chat_id=ctx.msg.chat_id, + channel=ctx.msg.channel, + chat_id=ctx.msg.chat_id, content="New session started.", ) +async def cmd_skill_list(ctx: CommandContext) -> OutboundMessage: + """List all available skills.""" + loader = ctx.loop.context.skills + skills = loader.list_skills(filter_unavailable=False) + if not skills: + return OutboundMessage( + channel=ctx.msg.channel, + chat_id=ctx.msg.chat_id, + content="No skills found.", + ) + lines = ["Available skills (use $ to activate):"] + for s in skills: + desc = loader._get_skill_description(s["name"]) + available = loader._check_requirements(loader._get_skill_meta(s["name"])) + mark = "✓" if available else "✗" + lines.append(f" {mark} {s['name']} — {desc}") + return OutboundMessage( + channel=ctx.msg.channel, + chat_id=ctx.msg.chat_id, + content="\n".join(lines), + metadata={"render_as": "text"}, + ) + + +async def intercept_skill_refs(ctx: CommandContext) -> OutboundMessage | None: + """Scan message for $skill-name references and inject matching skills.""" + refs = _SKILL_REF.findall(ctx.msg.content) + if not refs: + return None + loader = ctx.loop.context.skills + skill_names = {s["name"] for s in loader.list_skills(filter_unavailable=True)} + matched = [] + for name in dict.fromkeys(refs): # deduplicate, preserve order + if name in skill_names: + matched.append(name) + if not matched: + return None + # Strip matched $refs from the message + message = ctx.msg.content + for name in matched: + message = re.sub(rf"\${re.escape(name)}\b", "", message) + message = message.strip() + # Build injected content + skill_blocks = [] + for name in matched: + content = loader.load_skill(name) + if content: + stripped = loader._strip_frontmatter(content) + skill_blocks.append(f'\n{stripped}\n') + if not skill_blocks: + return None + names = ", ".join(f"'{n}'" for n in matched) + injected = ( + f"\n" + f"The user activated skill(s) {names} via $-reference. " + f"The following skill content was auto-appended by the system.\n" + + "\n".join(skill_blocks) + + "\n" + ) + ctx.msg.content = f"{injected}\n\n{message}" if message else injected + return None # fall through to LLM + + async def cmd_help(ctx: CommandContext) -> OutboundMessage: """Return available slash commands.""" return OutboundMessage( @@ -100,6 +170,8 @@ def build_help_text() -> str: "/stop — Stop the current task", "/restart — Restart the bot", "/status — Show bot status", + "/skills — List available skills", + "$ — Activate a skill inline (e.g. $weather what's the forecast)", "/help — Show available commands", ] return "\n".join(lines) @@ -113,3 +185,5 @@ def register_builtin_commands(router: CommandRouter) -> None: router.exact("/new", cmd_new) router.exact("/status", cmd_status) router.exact("/help", cmd_help) + router.exact("/skills", cmd_skill_list) + router.intercept(intercept_skill_refs) diff --git a/tests/cli/test_skill_command.py b/tests/cli/test_skill_command.py new file mode 100644 index 000000000..a322f648a --- /dev/null +++ b/tests/cli/test_skill_command.py @@ -0,0 +1,236 @@ +"""Tests for /skills listing and $skill inline activation.""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest + +from nanobot.bus.events import InboundMessage +from nanobot.command.builtin import cmd_skill_list, intercept_skill_refs +from nanobot.command.router import CommandContext + + +def _make_loop(): + """Create a minimal AgentLoop with mocked dependencies.""" + from nanobot.agent.loop import AgentLoop + from nanobot.bus.queue import MessageBus + + bus = MessageBus() + provider = MagicMock() + provider.get_default_model.return_value = "test-model" + workspace = MagicMock() + workspace.__truediv__ = MagicMock(return_value=MagicMock()) + + with ( + patch("nanobot.agent.loop.ContextBuilder"), + patch("nanobot.agent.loop.SessionManager"), + patch("nanobot.agent.loop.SubagentManager"), + ): + loop = AgentLoop(bus=bus, provider=provider, workspace=workspace) + return loop, bus + + +def _make_ctx(content: str, loop=None): + """Build a CommandContext for testing.""" + if loop is None: + loop, _ = _make_loop() + msg = InboundMessage(channel="cli", sender_id="user", chat_id="direct", content=content) + return CommandContext(msg=msg, session=None, key=msg.session_key, raw=content, loop=loop) + + +def _mock_skills_loader(skills=None, skill_content=None): + """Return a mock SkillsLoader with configurable data.""" + loader = MagicMock() + loader.list_skills.return_value = skills or [] + loader.load_skill.side_effect = lambda name: (skill_content or {}).get(name) + loader._get_skill_description.side_effect = lambda name: f"{name} description" + loader._get_skill_meta.return_value = {} + loader._check_requirements.return_value = True + loader._strip_frontmatter.side_effect = lambda c: c + return loader + + +WEATHER_SKILLS = [ + {"name": "weather", "path": "/skills/weather/SKILL.md", "source": "builtin"}, +] +MULTI_SKILLS = [ + {"name": "weather", "path": "/skills/weather/SKILL.md", "source": "builtin"}, + {"name": "github", "path": "/skills/github/SKILL.md", "source": "builtin"}, +] + + +class TestSkillList: + @pytest.mark.asyncio + async def test_lists_available_skills(self): + loop, _ = _make_loop() + loader = _mock_skills_loader(skills=MULTI_SKILLS) + loop.context = MagicMock() + loop.context.skills = loader + + ctx = _make_ctx("/skills", loop=loop) + result = await cmd_skill_list(ctx) + + assert result is not None + assert "weather" in result.content + assert "github" in result.content + assert "✓" in result.content + assert "$" in result.content # hints about $ usage + + @pytest.mark.asyncio + async def test_shows_unavailable_mark(self): + loop, _ = _make_loop() + loader = _mock_skills_loader( + skills=[{"name": "tmux", "path": "/skills/tmux/SKILL.md", "source": "builtin"}] + ) + loader._check_requirements.return_value = False + loop.context = MagicMock() + loop.context.skills = loader + + ctx = _make_ctx("/skills", loop=loop) + result = await cmd_skill_list(ctx) + + assert "✗" in result.content + assert "tmux" in result.content + + @pytest.mark.asyncio + async def test_no_skills(self): + loop, _ = _make_loop() + loader = _mock_skills_loader(skills=[]) + loop.context = MagicMock() + loop.context.skills = loader + + ctx = _make_ctx("/skills", loop=loop) + result = await cmd_skill_list(ctx) + + assert "No skills found" in result.content + + +class TestSkillInterceptor: + @pytest.mark.asyncio + async def test_injects_single_skill(self): + loop, _ = _make_loop() + loader = _mock_skills_loader( + skills=WEATHER_SKILLS, + skill_content={"weather": "Use the weather API."}, + ) + loop.context = MagicMock() + loop.context.skills = loader + + ctx = _make_ctx("$weather what is the forecast", loop=loop) + result = await intercept_skill_refs(ctx) + + assert result is None # falls through to LLM + assert '' in ctx.msg.content + assert "Use the weather API." in ctx.msg.content + assert "what is the forecast" in ctx.msg.content + + @pytest.mark.asyncio + async def test_injects_multiple_skills(self): + loop, _ = _make_loop() + loader = _mock_skills_loader( + skills=MULTI_SKILLS, + skill_content={ + "weather": "Weather skill content.", + "github": "GitHub skill content.", + }, + ) + loop.context = MagicMock() + loop.context.skills = loader + + ctx = _make_ctx("$weather $github do something", loop=loop) + result = await intercept_skill_refs(ctx) + + assert result is None + assert '' in ctx.msg.content + assert '' in ctx.msg.content + assert "do something" in ctx.msg.content + # Both skills wrapped in a single system-reminder + assert ctx.msg.content.count("") == 1 + + @pytest.mark.asyncio + async def test_skill_ref_anywhere_in_message(self): + loop, _ = _make_loop() + loader = _mock_skills_loader( + skills=WEATHER_SKILLS, + skill_content={"weather": "Weather skill content."}, + ) + loop.context = MagicMock() + loop.context.skills = loader + + ctx = _make_ctx("tell me $weather the forecast for NYC", loop=loop) + result = await intercept_skill_refs(ctx) + + assert result is None + assert '' in ctx.msg.content + assert ( + "tell me the forecast for NYC" in ctx.msg.content + or "tell me the forecast for NYC" in ctx.msg.content + ) + + @pytest.mark.asyncio + async def test_no_match_passes_through(self): + loop, _ = _make_loop() + loader = _mock_skills_loader(skills=WEATHER_SKILLS) + loop.context = MagicMock() + loop.context.skills = loader + + ctx = _make_ctx("just a normal message", loop=loop) + result = await intercept_skill_refs(ctx) + + assert result is None + assert ctx.msg.content == "just a normal message" + + @pytest.mark.asyncio + async def test_unknown_ref_ignored(self): + loop, _ = _make_loop() + loader = _mock_skills_loader(skills=WEATHER_SKILLS) + loop.context = MagicMock() + loop.context.skills = loader + + ctx = _make_ctx("$nonexistent do something", loop=loop) + result = await intercept_skill_refs(ctx) + + assert result is None + assert ctx.msg.content == "$nonexistent do something" + + @pytest.mark.asyncio + async def test_deduplicates_refs(self): + loop, _ = _make_loop() + loader = _mock_skills_loader( + skills=WEATHER_SKILLS, + skill_content={"weather": "Weather skill content."}, + ) + loop.context = MagicMock() + loop.context.skills = loader + + ctx = _make_ctx("$weather $weather forecast", loop=loop) + result = await intercept_skill_refs(ctx) + + assert result is None + assert ctx.msg.content.count('') == 1 + + @pytest.mark.asyncio + async def test_dollar_amount_not_matched(self): + loop, _ = _make_loop() + loader = _mock_skills_loader(skills=WEATHER_SKILLS) + loop.context = MagicMock() + loop.context.skills = loader + + ctx = _make_ctx("I have $100 in my account", loop=loop) + result = await intercept_skill_refs(ctx) + + assert result is None + assert ctx.msg.content == "I have $100 in my account" + + +class TestHelpIncludesSkill: + @pytest.mark.asyncio + async def test_help_shows_skill_commands(self): + loop, _ = _make_loop() + msg = InboundMessage(channel="cli", sender_id="user", chat_id="direct", content="/help") + response = await loop._process_message(msg) + + assert response is not None + assert "/skills" in response.content + assert "$" in response.content