From 7650a76a344bf9f63e4e30f6d9ad39e2986b6710 Mon Sep 17 00:00:00 2001 From: Perplexity Computer Date: Sat, 11 Apr 2026 00:40:52 +0000 Subject: [PATCH] feat(tools): add audit-backed dispatch router for agent-to-task routing Force-multiplier PR: routes tasks to agents based on historical merge rates and category strengths from the Perplexity contribution audit. Key data points baked in: - Timmy: 95% merge rate, 1.1 avg iterations (CI/CD, infra, security, hotfix) - Gemini: 88% merge rate, feature architecture + sovereign design - Allegro: 100% merge rate, best reviewer, infra + docs - Rockachopa: 44% merge rate, sovereign design in fleet-ops - Claude: 82% merge rate, frontend 3D + features (nexus only) Usage: python dispatch_router.py "fix: broken CI pipeline" fleet-ops Outputs JSON with recommended agent, score, reviewer, and all candidates. --- tools/dispatch_router.py | 228 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 228 insertions(+) create mode 100644 tools/dispatch_router.py diff --git a/tools/dispatch_router.py b/tools/dispatch_router.py new file mode 100644 index 00000000..9a82f12e --- /dev/null +++ b/tools/dispatch_router.py @@ -0,0 +1,228 @@ +#!/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()