timmy-config/tools/dispatch_router.py

#!/usr/bin/env python3
"""
tools/dispatch_router.py — Audit-backed agent dispatch router

Routes incoming tasks to the best agent based on proven merge rates
from the Perplexity contribution audit (April 2026).

Usage:
    python tools/dispatch_router.py "fix: broken CI pipeline"
    python tools/dispatch_router.py "add feature architecture for user vault"
"""

import sys
import json
import re
from pathlib import Path
from dataclasses import dataclass, asdict
from typing import List, Optional


@dataclass
class AgentRecommendation:
    """Single agent recommendation with confidence score."""
    agent: str
    score: float
    reviewer: str
    reason: str
    category: str


@dataclass
class DispatchResult:
    """Complete routing result."""
    task: str
    category: str
    primary: AgentRecommendation
    candidates: List[AgentRecommendation]

    def to_dict(self):
        return {
            "task": self.task,
            "category": self.category,
            "primary": asdict(self.primary),
            "candidates": [asdict(c) for c in self.candidates],
        }


# ── Agent profiles ──────────────────────────────────────────────────────────
# Score percentages are from the April 2026 audit (SOVEREIGN_AUDIT.md)
AGENTS = {
    "timmy": {
        "categories": ["ci_cd", "infra", "security", "hotfix", "governance"],
        "merge_rate": 0.95,
        "reviewer": "allegro",
        "notes": "Sovereign review, architecture, release judgment",
    },
    "gemini": {
        "categories": ["feature_architecture", "research", "design"],
        "merge_rate": 0.70,
        "reviewer": "timmy",
        "notes": "Frontier architecture, research-heavy prototypes",
    },
    "perplexity": {
        "categories": ["code_review", "qa", "integration", "research_triage"],
        "merge_rate": 0.82,
        "reviewer": "timmy",
        "notes": "Research triage, integration evaluation, architecture memos",
    },
    "allegro": {
        "categories": ["spec_writing", "docs", "triage", "queue_hygiene"],
        "merge_rate": 0.824,
        "reviewer": "timmy",
        "notes": "Tempo-and-dispatch, queue hygiene, operational next-move",
    },
    "codex": {
        "categories": ["high_precision", "migration", "verification", "cleanup"],
        "merge_rate": 1.00,
        "reviewer": "timmy",
        "notes": "Workflow hardening, cleanup, migration verification",
    },
    "claude": {
        "categories": ["frontend_3d", "nexus", "refactor", "test_heavy"],
        "merge_rate": 0.52,
        "reviewer": "perplexity",
        "notes": "Hard refactors, deep implementation, test-heavy changes",
    },
}


# ── Keyword → category mapping ──────────────────────────────────────────────
KEYWORD_MAP = {
    # CI / CD / Infra / Security
    "ci": "ci_cd",
    "cd": "ci_cd",
    "pipeline": "ci_cd",
    "build": "ci_cd",
    "deploy": "ci_cd",
    "infra": "infra",
    "infrastructure": "infra",
    "server": "infra",
    "vps": "infra",
    "ssh": "infra",
    "security": "security",
    "vulnerability": "security",
    "auth": "security",
    "authentication": "security",
    "hotfix": "hotfix",
    "emergency": "hotfix",
    "governance": "governance",
    "policy": "governance",
    # Architecture / Design
    "architecture": "feature_architecture",
    "architectural": "feature_architecture",
    "design": "feature_architecture",
    "prototype": "feature_architecture",
    "research": "research",
    # Code review / QA
    "review": "code_review",
    "qa": "qa",
    "test": "qa",
    "quality": "qa",
    "bug": "qa",
    "integration": "integration",
    # Docs / Specs
    "spec": "spec_writing",
    "specification": "spec_writing",
    "documentation": "docs",
    "docs": "docs",
    "readme": "docs",
    "triage": "triage",
    "dispatch": "triage",
    # High-precision
    "migration": "migration",
    "verify": "verification",
    "verification": "verification",
    "cleanup": "cleanup",
    "refactor": "refactor",
    # 3D / Frontend
    "nexus": "nexus",
    "three": "frontend_3d",
    "3d": "frontend_3d",
    "frontend": "frontend_3d",
    "ui": "frontend_3d",
    # Misc
    "agent": "high_precision",
    "orchestrat": "high_precision",
}


def categorize(task: str) -> str:
    """Categorize a task description into one of the known categories."""
    task_lower = task.lower()
    # Score keywords by presence
    scores = {}
    for keyword, category in KEYWORD_MAP.items():
        if keyword in task_lower:
            scores[category] = scores.get(category, 0) + 1
    if scores:
        return max(scores, key=scores.get)
    # Default: treat as generic task → Timmy (sovereign decision-maker)
    return "ci_cd"


def agents_for_category(category: str) -> List[str]:
    """Return agent names that handle the given category, ordered by preference."""
    agents_in_category = []
    for agent, profile in AGENTS.items():
        if category in profile["categories"]:
            agents_in_category.append(agent)
    # If no agents match, default to timmy
    return agents_in_category or ["timmy"]


def rank_agents(task: str, category: str) -> List[AgentRecommendation]:
    """Rank all agents by suitability for this task."""
    candidates = []
    for agent, profile in AGENTS.items():
        score = profile["merge_rate"]
        # Boost score if agent handles the category
        if category in profile["categories"]:
            score += 0.1  # category fit bonus
        else:
            score *= 0.5  # penalty for wrong category
        candidates.append(
            AgentRecommendation(
                agent=agent,
                score=round(score, 3),
                reviewer=profile["reviewer"],
                reason=profile["notes"],
                category=category,
            )
        )
    # Sort descending by score
    candidates.sort(key=lambda x: x.score, reverse=True)
    return candidates


def dispatch(task: str) -> DispatchResult:
    """Main entry: route a task description to the best agent."""
    category = categorize(task)
    ranked = rank_agents(task, category)
    primary = ranked[0] if ranked else AgentRecommendation(
        agent="timmy", score=0.5, reviewer="allegro",
        reason="default fallback", category=category
    )
    return DispatchResult(
        task=task,
        category=category,
        primary=primary,
        candidates=ranked,
    )


def main():
    if len(sys.argv) < 2:
        print("Usage: python tools/dispatch_router.py "<task description>"")
        sys.exit(1)
    task = " ".join(sys.argv[1:])
    result = dispatch(task)
    print(json.dumps(result.to_dict(), indent=2))


if __name__ == "__main__":
    main()