feat(phase33): custom skills, workflow recipes, and dynamic slash command autocompletion

agentcli/agent/registry.py

@@ -31,6 +31,7 @@
 from ..subagents.file_ops import FileOpsAgent
 from ..subagents.grep_search import GrepSearchAgent
 from ..subagents.shell import ShellExecutionAgent
+from ..subagents.skill_runner import SkillRunnerAgent
 from ..subagents.task_manager import TaskManagerAgent
 from ..subagents.web_fetch import WebFetchAgent
 from ..subagents.web_search import WebSearchAgent
@@ -230,6 +231,10 @@ def _register_defaults(self) -> None:
         self.register(
             SubAgentType.DIAGNOSTICS_CHECK.value, lambda: DiagnosticsAgent(config=diag_cfg)
         )
+        self.register(
+            SubAgentType.SKILL_RUNNER.value, lambda: SkillRunnerAgent()
+        )
+
 
     @staticmethod
     def _safe_type(agent_type: str) -> SubAgentType:

agentcli/cli.py

@@ -922,7 +922,102 @@ async def run_chat(args: argparse.Namespace, config: Config) -> int:
                     print("Usage: /tasks [list | status <id> | logs <id> | kill <id>]")
                 continue
 
+            if user_input.startswith(("/skill", "/skills")):
+                skill_parts = user_input.split(maxsplit=2)
+                skill_subcmd = skill_parts[1].lower() if len(skill_parts) > 1 else "list"
+                target_skill = skill_parts[2].strip() if len(skill_parts) > 2 else ""
+
+                if skill_subcmd in ("list", "ls") or len(skill_parts) == 1:
+                    skills_list = session.skill_engine.list_available_skills()
+                    if not skills_list:
+                        print("No skills discovered in workspace or user directory.")
+                    else:
+                        print(f"\n--- Available Skills ({len(skills_list)}) ---")
+                        for s in skills_list:
+                            source_badge = f"[{s['source_type'].upper()}]"
+                            print(f"  {source_badge} {s['name']} (v{s['version']}): {s['description']}")
+                        print("------------------------------------\n")
+                        print("Run a skill with: /skill run <name> or inspect with /skill info <name>\n")
+                elif skill_subcmd == "info":
+                    if not target_skill:
+                        print("Usage: /skill info <skill_name>")
+                    else:
+                        manifest = session.skill_engine.loader.get_skill(target_skill)
+                        if not manifest:
+                            print(f"Skill '{target_skill}' not found. Run '/skill list' to see available skills.")
+                        else:
+                            print(f"\nSkill [{manifest.name}] (v{manifest.version})")
+                            print(f"  Description: {manifest.description}")
+                            print(f"  Source: {manifest.source_type} ({manifest.skill_dir or 'builtin'})")
+                            print(f"  Execution Mode: {manifest.execution_mode} (max {manifest.max_iterations} iterations)")
+                            if manifest.parameters:
+                                print("  Parameters:")
+                                for p_name, p in manifest.parameters.items():
+                                    req_str = " (required)" if p.required else f" (default: {p.default})"
+                                    print(f"    - {p_name} [{p.type}]{req_str}: {p.description}")
+                            if manifest.required_tools:
+                                print(f"  Required Tools: {', '.join(manifest.required_tools)}")
+                            print()
+                elif skill_subcmd == "reload":
+                    session.skill_engine.loader.reload()
+                    total = len(session.skill_engine.loader.list_skills())
+                    print(f"Reloaded skills. Found {total} skill(s).")
+                elif skill_subcmd in ("run", "exec") or session.skill_engine.loader.has_skill(skill_subcmd):
+                    if skill_subcmd in ("run", "exec"):
+                        actual_name = target_skill.split(maxsplit=1)[0] if target_skill else ""
+                        raw_args_str = target_skill.split(maxsplit=1)[1] if len(target_skill.split(maxsplit=1)) > 1 else ""
+                    else:
+                        actual_name = skill_subcmd
+                        raw_args_str = target_skill
+
+                    if not actual_name:
+                        print("Usage: /skill run <skill_name> [param=value ...]")
+                        continue
+
+                    parsed_args: dict[str, Any] = {}
+                    if raw_args_str:
+                        for token in raw_args_str.split():
+                            if "=" in token:
+                                k, v = token.split("=", 1)
+                                parsed_args[k.strip()] = v.strip()
+                            else:
+                                parsed_args["target"] = token
+
+                    try:
+                        manifest, rendered = session.skill_engine.prepare_skill(
+                            skill_name=actual_name,
+                            arguments=parsed_args,
+                        )
+                        print(f"\nExecuting skill '{manifest.name}' ({manifest.execution_mode})...\n")
+
+                        if manifest.execution_mode == "loop" or session.should_use_loop(rendered):
+                            await session.async_add_user_message(rendered)
+                            loop_summary = None
+                            try:
+                                with renderer.status_spinner(f"Running skill {manifest.name}..."):
+                                    async for event in session.run_loop(rendered):
+                                        renderer.render_loop_event(event, verbose=verbose)
+                                        if isinstance(event, FinishEvent):
+                                            loop_summary = event.summary
+                                            if getattr(event, "output", None):
+                                                loop_summary = f"{event.summary}\n\n{event.output}"
+                                        elif isinstance(event, LoopErrorEvent):
+                                            loop_summary = f"[loop error] {event.error}"
+                                await session.async_add_assistant_message(
+                                    f"[Skill: {manifest.name}]\n{loop_summary or '(completed)'}"
+                                )
+                            except LoopIterationLimitError as exc:
+                                print(f"\n[skill-runner] Iteration limit reached: {exc}")
+                            except KeyboardInterrupt:
+                                print("\n[interrupted]")
+                    except KeyError as k_err:
+                        print(f"Error: {k_err}")
+                else:
+                    print("Usage: /skill [list | info <name> | run <name> [args] | reload]")
+                continue
+
             if user_input in {"/clear", "/cls"}:
+
                 renderer.clear()
                 continue
 
@@ -938,7 +1033,7 @@ async def run_chat(args: argparse.Namespace, config: Config) -> int:
                     print("Usage: /goal <task description>")
                     continue
                 goal_text = parts[1].strip()
-                loop_summary: str | None = None
+                loop_summary = None
                 try:
                     with renderer.status_spinner("Running autonomous goal loop..."):
                         async for event in session.run_loop(goal_text):

agentcli/session.py

@@ -25,10 +25,12 @@
 from .routing.classifier import classify
 from .routing.registry import ModelRegistry
 from .routing.router import Router
+from .skills.engine import SkillEngine
 
 logger = logging.getLogger(__name__)
 
 
+
 @dataclass
 class SessionReply:
     stream: AsyncIterator[str]
@@ -111,6 +113,8 @@ def __init__(
         self.mcp_manager: MCPClientManager = MCPClientManager(config=self.config)
         self.checkpoint_manager: CheckpointManager = CheckpointManager()
         self.task_manager: TaskManager = TaskManager()
+        self.skill_engine: SkillEngine = SkillEngine()
+
 
         if config.routing.enabled:
             self.registry = ModelRegistry(config.routing)

agentcli/skills/__init__.py

@@ -0,0 +1,15 @@
+"""Skills and custom workflow recipe system for agentcli (Phase 33)."""
+
+from .engine import SkillEngine, SkillExecutionResult
+from .loader import SkillLoader
+from .manifest import SkillManifest, SkillParameter, SkillStep, parse_skill_markdown
+
+__all__ = [
+    "SkillEngine",
+    "SkillExecutionResult",
+    "SkillLoader",
+    "SkillManifest",
+    "SkillParameter",
+    "SkillStep",
+    "parse_skill_markdown",
+]

agentcli/skills/builtin/code_review/SKILL.md

@@ -0,0 +1,40 @@
+---
+name: code-review
+description: Comprehensive code review analyzing logic, security vulnerabilities, edge cases, and design patterns
+version: 1.0.0
+author: agentcli
+execution_mode: loop
+max_iterations: 5
+parameters:
+  target:
+    type: string
+    default: "."
+    description: File or directory path to review
+  focus:
+    type: enum
+    options: [all, security, performance, correctness, architecture, style]
+    default: all
+    description: Specific review focus area
+required_tools: [grep_search, file_ops, diagnostics_check]
+---
+# Autonomous Code Review
+
+Please perform a thorough, actionable code review for the target: `{{target}}`.
+
+## Review Focus
+- **Active Focus Area**: `{{focus}}`
+- **Active Git Branch**: `{{active_branch}}`
+- **Workspace Directory**: `{{workspace_dir}}`
+
+## Step-by-Step Review Instructions
+1. Inspect the target path `{{target}}` using `file_ops` and `grep_search`.
+2. Run any project diagnostics using `diagnostics_check` to detect underlying linter or type defects.
+3. Review logic for:
+   - **Correctness**: Off-by-one errors, unhandled edge cases, null/none dereferences.
+   - **Security**: Injection risks, sensitive credential leaks, insecure deserialization, unsafe shell executions.
+   - **Performance**: High algorithmic complexity, unindexed searches, unclosed handles or locks.
+   - **Architecture & Modularity**: Adherence to project conventions and single-responsibility principles.
+4. Output a structured findings report:
+   - **Critical Issues** (Immediate attention required)
+   - **Warnings & Improvements** (Maintainability & performance enhancements)
+   - **Code Examples** (Precise before/after refactor snippets)

agentcli/skills/builtin/security_audit/SKILL.md

@@ -0,0 +1,37 @@
+---
+name: security-audit
+description: Autonomous vulnerability, secret leak, and dependency supply-chain security audit
+version: 1.0.0
+author: agentcli
+execution_mode: loop
+max_iterations: 5
+parameters:
+  target_dir:
+    type: string
+    default: "."
+    description: Directory path to audit
+  scan_secrets:
+    type: bool
+    default: true
+    description: Scan for hardcoded API keys, JWT tokens, and private credentials
+required_tools: [grep_search, file_ops, diagnostics_check]
+---
+# Security & Vulnerability Audit
+
+Perform an end-to-end security review of directory: `{{target_dir}}`.
+
+## Parameters
+- **Target Directory**: `{{target_dir}}`
+- **Scan Secrets**: `{{scan_secrets}}`
+- **Active Branch**: `{{active_branch}}`
+
+## Audit Checklist
+1. **Secret & Key Leaks**:
+   - Search for API keys, private keys, JWT secrets, passwords, or tokens in committed source code or configuration.
+2. **Command & SQL/NoSQL Injection**:
+   - Inspect all raw shell commands, formatted strings in subprocess execution, or unsanitized DB queries.
+3. **Authentication & Authorization**:
+   - Verify proper JWT token verification, RBAC scoping, and tenant isolation boundaries.
+4. **Input Sanitization & Path Traversal**:
+   - Check file read/write paths for `../` path traversal vulnerabilities.
+5. Provide a prioritized markdown vulnerability table (Severity, File, Line, Vulnerability, Remediation).

agentcli/skills/builtin/test_generator/SKILL.md

@@ -0,0 +1,41 @@
+---
+name: test-generator
+description: Autonomous unit and integration test generator for target modules with edge-case coverage
+version: 1.0.0
+author: agentcli
+execution_mode: loop
+max_iterations: 6
+parameters:
+  target_file:
+    type: string
+    default: "."
+    description: Target source file or module to generate tests for
+  framework:
+    type: enum
+    options: [pytest, unittest, vitest, jest, cargo]
+    default: pytest
+    description: Testing framework
+  edge_cases:
+    type: bool
+    default: true
+    description: Whether to explicitly generate extreme edge-case and boundary tests
+required_tools: [grep_search, file_ops, shell_execution, diagnostics_check]
+---
+# Autonomous Test Generator
+
+Generate comprehensive, hermetic tests for target: `{{target_file}}` using framework: `{{framework}}`.
+
+## Parameters
+- **Target File**: `{{target_file}}`
+- **Framework**: `{{framework}}`
+- **Include Edge Cases**: `{{edge_cases}}`
+- **Active Workspace**: `{{workspace_dir}}`
+
+## Instructions
+1. Read `{{target_file}}` and understand all public methods, classes, signatures, and internal error branches.
+2. Locate existing project test suites to follow naming conventions, fixtures, and directory structures.
+3. Write isolated unit and integration test functions covering:
+   - Happy path behaviors.
+   - Boundary values, empty inputs, invalid types, and unexpected formats.
+   - Async coroutines, exception handling, and teardown cleanup.
+4. Execute test suite and verify 100% pass rate using `shell_execution` or `diagnostics_check`.

agentcli/skills/engine.py

@@ -0,0 +1,82 @@
+"""Skill execution engine and recipe orchestrator for Phase 33."""
+
+from __future__ import annotations
+
+import logging
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from .loader import SkillLoader
+from .manifest import SkillManifest
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class SkillExecutionResult:
+    """Result emitted after executing a skill."""
+
+    success: bool
+    skill_name: str
+    rendered_prompt: str
+    output: Any = None
+    error: str | None = None
+    execution_mode: str = "loop"
+    tools_used: list[str] | None = None
+
+
+class SkillEngine:
+    """Coordinates skill discovery, validation, context injection, and execution."""
+
+    def __init__(
+        self,
+        loader: SkillLoader | None = None,
+        workspace_dir: str | Path | None = None,
+    ) -> None:
+        self.workspace_dir = Path(workspace_dir).resolve() if workspace_dir else Path.cwd()
+        self.loader = loader or SkillLoader(workspace_dir=self.workspace_dir)
+
+    def _get_git_branch(self) -> str:
+        """Get the current active git branch name, if any."""
+        try:
+            res = subprocess.run(
+                ["git", "rev-parse", "--abbrev-ref", "HEAD"],
+                cwd=str(self.workspace_dir),
+                capture_output=True,
+                text=True,
+                check=False,
+            )
+            if res.returncode == 0:
+                return res.stdout.strip()
+        except Exception as exc:  # noqa: BLE001
+            logger.debug("Could not determine git branch for workspace %s: %s", self.workspace_dir, exc)
+        return "main"
+
+    def prepare_skill(
+        self,
+        skill_name: str,
+        arguments: dict[str, Any] | None = None,
+        extra_context: dict[str, Any] | None = None,
+    ) -> tuple[SkillManifest, str]:
+        """Validate arguments and render prompt for a skill."""
+        manifest = self.loader.get_skill(skill_name)
+        if not manifest:
+            raise KeyError(f"Skill '{skill_name}' not found. Run '/skill list' to see available skills.")
+
+        args = arguments or {}
+        context: dict[str, Any] = {
+            "workspace_dir": str(self.workspace_dir),
+            "workspace_name": self.workspace_dir.name,
+            "active_branch": self._get_git_branch(),
+        }
+        if extra_context:
+            context.update(extra_context)
+
+        rendered_prompt = manifest.render_prompt(args, context=context)
+        return manifest, rendered_prompt
+
+    def list_available_skills(self) -> list[dict[str, Any]]:
+        """Return a serializable list of all loaded skill summaries."""
+        return [skill.to_dict() for skill in self.loader.list_skills()]