#!/usr/bin/env python3 """Dispatch Router — data-driven agent-to-task routing. Multiplies the fleet's best force factor (Timmy: 95%+ merge rate across all repos) by routing every new issue/task to the agent with the highest historical success rate for that task category. Data source: Perplexity contribution audit 2026-04-10 """ import json import sys from dataclasses import dataclass, field from enum import Enum from typing import Optional class TaskCategory(Enum): CI_CD = "ci_cd" INFRA_ANSIBLE = "infra_ansible" SECURITY = "security" FEATURE_ARCH = "feature_architecture" FRONTEND_3D = "frontend_3d" DOCS = "docs" SOVEREIGN_DESIGN = "sovereign_design" REVIEW = "review" HOTFIX = "hotfix" RESEARCH = "research" @dataclass class AgentProfile: name: str strengths: list[TaskCategory] repos: list[str] merge_rate: float # 0.0-1.0 avg_iterations: float # avg PRs before merge can_review: bool = False # ---- Audit-backed agent registry (2026-04-10 snapshot) ---- AGENT_REGISTRY: dict[str, AgentProfile] = { "timmy": AgentProfile( name="Timmy", strengths=[ TaskCategory.CI_CD, TaskCategory.INFRA_ANSIBLE, TaskCategory.SECURITY, TaskCategory.HOTFIX, ], repos=["the-nexus", "timmy-config", "fleet-ops", "the-beacon"], merge_rate=0.95, avg_iterations=1.1, ), "gemini": AgentProfile( name="Gemini", strengths=[ TaskCategory.FEATURE_ARCH, TaskCategory.SOVEREIGN_DESIGN, TaskCategory.DOCS, ], repos=["the-nexus", "timmy-config", "fleet-ops"], merge_rate=0.88, avg_iterations=1.2, can_review=True, ), "allegro": AgentProfile( name="Allegro", strengths=[ TaskCategory.REVIEW, TaskCategory.INFRA_ANSIBLE, TaskCategory.DOCS, ], repos=["the-nexus", "fleet-ops"], merge_rate=1.0, avg_iterations=1.0, can_review=True, ), "rockachopa": AgentProfile( name="Rockachopa", strengths=[ TaskCategory.SOVEREIGN_DESIGN, TaskCategory.RESEARCH, TaskCategory.FRONTEND_3D, ], repos=["the-nexus", "fleet-ops"], merge_rate=0.44, avg_iterations=2.8, ), "claude": AgentProfile( name="Claude", strengths=[ TaskCategory.FRONTEND_3D, TaskCategory.FEATURE_ARCH, ], repos=["the-nexus"], merge_rate=0.82, avg_iterations=1.3, ), } # ---- Keyword to category mapping ---- KEYWORD_MAP: dict[str, TaskCategory] = { "ci": TaskCategory.CI_CD, "pipeline": TaskCategory.CI_CD, "lint": TaskCategory.CI_CD, "workflow": TaskCategory.CI_CD, "ansible": TaskCategory.INFRA_ANSIBLE, "role": TaskCategory.INFRA_ANSIBLE, "playbook": TaskCategory.INFRA_ANSIBLE, "deploy": TaskCategory.INFRA_ANSIBLE, "nginx": TaskCategory.INFRA_ANSIBLE, "vault": TaskCategory.SECURITY, "secret": TaskCategory.SECURITY, "ssh": TaskCategory.SECURITY, "gitignore": TaskCategory.SECURITY, "feat": TaskCategory.FEATURE_ARCH, "bridge": TaskCategory.FEATURE_ARCH, "integration": TaskCategory.FEATURE_ARCH, "api": TaskCategory.FEATURE_ARCH, "mnemosyne": TaskCategory.FRONTEND_3D, "3d": TaskCategory.FRONTEND_3D, "holographic": TaskCategory.FRONTEND_3D, "orb": TaskCategory.FRONTEND_3D, "spatial": TaskCategory.FRONTEND_3D, "doc": TaskCategory.DOCS, "manual": TaskCategory.DOCS, "readme": TaskCategory.DOCS, "sovereign": TaskCategory.SOVEREIGN_DESIGN, "alignment": TaskCategory.SOVEREIGN_DESIGN, "governance": TaskCategory.SOVEREIGN_DESIGN, "fix": TaskCategory.HOTFIX, "bug": TaskCategory.HOTFIX, "broken": TaskCategory.HOTFIX, "crash": TaskCategory.HOTFIX, "audit": TaskCategory.RESEARCH, "research": TaskCategory.RESEARCH, "investigate": TaskCategory.RESEARCH, } def classify_task(title: str, labels: Optional[list[str]] = None) -> TaskCategory: """Classify a task by scanning title + labels for keywords.""" text = title.lower() if labels: text += " " + " ".join(l.lower() for l in labels) for keyword, category in KEYWORD_MAP.items(): if keyword in text: return category return TaskCategory.FEATURE_ARCH # default def route( title: str, repo: str, labels: Optional[list[str]] = None, ) -> list[dict]: """Return ranked agent recommendations for a task. Returns list of {agent, score, reason} sorted best-first. """ category = classify_task(title, labels) candidates = [] for agent_id, profile in AGENT_REGISTRY.items(): if repo not in profile.repos: continue score = profile.merge_rate * 100 reason_parts = [] if category in profile.strengths: score += 30 reason_parts.append(f"strength: {category.value}") # Penalise high-iteration agents score -= (profile.avg_iterations - 1.0) * 10 reason_parts.append(f"merge_rate={profile.merge_rate:.0%}") reason_parts.append(f"avg_iter={profile.avg_iterations:.1f}") candidates.append( { "agent": profile.name, "score": round(score, 1), "category": category.value, "reason": ", ".join(reason_parts), } ) candidates.sort(key=lambda c: c["score"], reverse=True) return candidates def pick_reviewer(exclude: str, repo: str) -> Optional[str]: """Pick the best reviewer who isn't the author.""" for agent_id, profile in AGENT_REGISTRY.items(): if agent_id == exclude.lower(): continue if not profile.can_review: continue if repo in profile.repos: return profile.name return None def main(): """CLI: dispatch_router.py <repo> [label1,label2,...]""" if len(sys.argv) < 3: print("Usage: dispatch_router.py <title> <repo> [labels]") sys.exit(1) title = sys.argv[1] repo = sys.argv[2] labels = sys.argv[3].split(",") if len(sys.argv) > 3 else None result = route(title, repo, labels) if not result: print(json.dumps({"error": "no candidates for repo", "repo": repo})) sys.exit(1) best = result[0] reviewer = pick_reviewer(best["agent"], repo) output = { "recommended_agent": best["agent"], "score": best["score"], "category": best["category"], "reason": best["reason"], "reviewer": reviewer, "all_candidates": result, } print(json.dumps(output, indent=2)) if __name__ == "__main__": main()