scripts-skill-selector

#!/usr/bin/env python3 """

title: Intelligent Skill and Agent Selector component_type: script version: 1.0.0 audience: contributor status: active summary: Queries component database to find optimal skill/agent for a task keywords:

skill
agent
selection
routing
database created: 2026-01-03 updated: 2026-01-03 script_name: skill-selector.py language: python executable: true usage: python3 scripts/skill-selector.py "task description"

Intelligent Skill and Agent Selector

Queries the component database (platform.db - ADR-118 Tier 1) to find the best skill, command, or agent for a given task. Emits structured output with success markers for retrospective scoring.

Usage: python3 scripts/skill-selector.py "deploy to kubernetes" python3 scripts/skill-selector.py --type agent "write documentation" python3 scripts/skill-selector.py --type skill "analyze code quality" python3 scripts/skill-selector.py --all "orchestrate multi-step task" """

import argparse import json import sqlite3 import sys from pathlib import Path from typing import Dict, List, Optional, Tuple

Root directory

SCRIPT_DIR = Path(file).parent ROOT_DIR = SCRIPT_DIR.parent

ADR-114 & ADR-118: Use centralized path discovery

sys.path.insert(0, str(SCRIPT_DIR / "core")) try: from paths import get_platform_db_path, PLATFORM_DB DB_PATH = PLATFORM_DB # Component data goes to platform.db (Tier 1) except ImportError: # Fallback for backward compatibility _user_data = Path.home() / "PROJECTS" / ".coditect-data" / "context-storage" if _user_data.exists(): PLATFORM_DB = _user_data / "platform.db" else: PLATFORM_DB = Path.home() / ".coditect" / "context-storage" / "platform.db" DB_PATH = PLATFORM_DB # Backward compatibility alias

class SkillSelector: """Query component database for optimal skill/agent selection."""

def __init__(self, db_path: Path = DB_PATH):
    self.db_path = db_path
    self.conn = None

def connect(self) -> bool:
    """Connect to the component database."""
    if not self.db_path.exists():
        print(f"❌ ERROR: Database not found at {self.db_path}")
        print("  Run: python3 scripts/component-indexer.py")
        return False

    try:
        self.conn = sqlite3.connect(str(self.db_path))
        self.conn.row_factory = sqlite3.Row
        return True
    except Exception as e:
        print(f"❌ ERROR: Failed to connect to database: {e}")
        return False

def search(
    self,
    query: str,
    component_type: Optional[str] = None,
    limit: int = 5
) -> List[Dict]:
    """Search for components matching the query."""
    if not self.conn:
        if not self.connect():
            return []

    results = []

    # Full-text search query
    sql = """
        SELECT
            c.name,
            c.type,
            c.path,
            c.description,
            c.confidence,
            c.llm_model,
            c.invocation_method
        FROM component_search cs
        JOIN components c ON cs.id = c.id
        WHERE component_search MATCH ?
    """

    # FTS5 requires special handling for multi-word queries
    # Convert "deploy to kubernetes" to "deploy OR to OR kubernetes"
    words = query.split()
    if len(words) > 1:
        fts_query = " OR ".join(words)
    else:
        fts_query = query

    params = [fts_query]

    if component_type:
        sql += " AND c.type = ?"
        params.append(component_type)

    sql += " ORDER BY rank LIMIT ?"
    params.append(limit)

    try:
        cursor = self.conn.execute(sql, params)
        for row in cursor:
            results.append({
                "name": row["name"],
                "type": row["type"],
                "path": row["path"],
                "description": row["description"][:200] if row["description"] else "",
                "confidence": row["confidence"],
                "model": row["llm_model"],
                "invocation": row["invocation_method"]
            })
    except Exception as e:
        # Fallback to LIKE search if FTS fails
        sql = """
            SELECT name, type, path, description, confidence, llm_model, invocation_method
            FROM components
            WHERE (name LIKE ? OR description LIKE ?)
        """
        params = [f"%{query}%", f"%{query}%"]

        if component_type:
            sql += " AND type = ?"
            params.append(component_type)

        sql += " ORDER BY confidence DESC LIMIT ?"
        params.append(limit)

        cursor = self.conn.execute(sql, params)
        for row in cursor:
            results.append({
                "name": row["name"],
                "type": row["type"],
                "path": row["path"],
                "description": row["description"][:200] if row["description"] else "",
                "confidence": row["confidence"],
                "model": row["llm_model"],
                "invocation": row["invocation_method"]
            })

    return results

def get_capabilities(self, component_name: str) -> List[str]:
    """Get capabilities for a component."""
    if not self.conn:
        return []

    sql = """
        SELECT cap.capability
        FROM capabilities cap
        JOIN components c ON cap.component_id = c.id
        WHERE c.name = ?
        LIMIT 10
    """

    try:
        cursor = self.conn.execute(sql, [component_name])
        return [row[0] for row in cursor]
    except:
        return []

def recommend(
    self,
    task: str,
    component_type: Optional[str] = None
) -> Dict:
    """Get recommendation for a task with structured output."""

    # Search all types or specific type
    types_to_search = [component_type] if component_type else ["agent", "skill", "command"]

    all_results = []
    for ctype in types_to_search:
        results = self.search(task, ctype, limit=3)
        all_results.extend(results)

    # Sort by confidence
    all_results.sort(key=lambda x: x.get("confidence", 0), reverse=True)

    if not all_results:
        return {
            "success": False,
            "error": f"No components found for task: {task}",
            "recommendation": None
        }

    primary = all_results[0]
    alternatives = all_results[1:4] if len(all_results) > 1 else []

    # Get capabilities for primary
    capabilities = self.get_capabilities(primary["name"])

    # Build invocation string
    if primary["type"] == "agent":
        if primary.get("invocation"):
            invocation = primary["invocation"]
        else:
            invocation = f'Task(subagent_type="{primary["name"]}", prompt="{task}")'
    elif primary["type"] == "command":
        invocation = f'/{primary["name"]} {task}'
    elif primary["type"] == "skill":
        invocation = f'Use {primary["name"]} skill for: {task}'
    else:
        invocation = f'{primary["type"]}: {primary["name"]}'

    return {
        "success": True,
        "task": task,
        "recommendation": {
            "name": primary["name"],
            "type": primary["type"],
            "confidence": primary["confidence"],
            "description": primary["description"],
            "invocation": invocation,
            "capabilities": capabilities[:5]
        },
        "alternatives": [
            {"name": a["name"], "type": a["type"], "confidence": a["confidence"]}
            for a in alternatives
        ]
    }

def main(): parser = argparse.ArgumentParser( description="Intelligent skill/agent selector" ) parser.add_argument("task", nargs="?", help="Task description") parser.add_argument("--type", "-t", choices=["agent", "skill", "command"], help="Filter by component type") parser.add_argument("--all", "-a", action="store_true", help="Search all component types") parser.add_argument("--json", "-j", action="store_true", help="Output as JSON") parser.add_argument("--quiet", "-q", action="store_true", help="Minimal output (just the invocation)")

args = parser.parse_args()

if not args.task:
    parser.print_help()
    sys.exit(1)

selector = SkillSelector()
result = selector.recommend(args.task, args.type)

if args.json:
    print(json.dumps(result, indent=2))
    sys.exit(0 if result["success"] else 1)

if args.quiet:
    if result["success"]:
        print(result["recommendation"]["invocation"])
    sys.exit(0 if result["success"] else 1)

# Formatted output with success markers
print()
print("=" * 60)
print("CODITECT Skill Selector")
print("=" * 60)
print(f"Task: {args.task}")
print()

if not result["success"]:
    print(f"❌ SELECTION FAILED: {result['error']}")
    sys.exit(1)

rec = result["recommendation"]
print(f"✅ PRIMARY RECOMMENDATION")
print(f"   Name: {rec['name']}")
print(f"   Type: {rec['type']}")
print(f"   Confidence: {rec['confidence']:.0%}")
print()
print(f"   Invocation:")
print(f"   {rec['invocation']}")
print()

if rec.get("capabilities"):
    print(f"   Capabilities:")
    for cap in rec["capabilities"]:
        print(f"   - {cap}")
    print()

if result.get("alternatives"):
    print("   Alternatives:")
    for alt in result["alternatives"]:
        print(f"   - {alt['name']} ({alt['type']}, {alt['confidence']:.0%})")
    print()

print("=" * 60)
print("✅ SKILL SELECTION COMPLETE")
print("=" * 60)

sys.exit(0)

if name == "main": main()

#!/usr/bin/env python3 """​

Root directory

ADR-114 & ADR-118: Use centralized path discovery

#!/usr/bin/env python3 """