Compare commits

...

1 Commits

Author SHA1 Message Date
d94016458a feat(tools): operationalize dispatch router into triage loop
Some checks failed
Architecture Lint / Linter Tests (pull_request) Successful in 23s
Smoke Test / smoke (pull_request) Failing after 19s
Validate Config / YAML Lint (pull_request) Failing after 14s
Validate Config / JSON Validate (pull_request) Successful in 16s
Validate Config / Python Syntax & Import Check (pull_request) Failing after 39s
Validate Config / Python Test Suite (pull_request) Has been skipped
Validate Config / Shell Script Lint (pull_request) Failing after 44s
Validate Config / Cron Syntax Check (pull_request) Successful in 10s
Validate Config / Deploy Script Dry Run (pull_request) Successful in 9s
Validate Config / Playbook Schema Validation (pull_request) Successful in 24s
PR Checklist / pr-checklist (pull_request) Failing after 3m45s
Architecture Lint / Lint Repository (pull_request) Failing after 20s
- Add tools/dispatch_router.py: audit-backed agent router based on
  April 2026 contribution audit (Timmy 95%, Gemini 70%, Perplexity 82%,
  Allegro 82.4%, Codex 100%, Claude 52%).
- Integrate into bin/timmy-orchestrator.sh: when AUTO_ASSIGN_UNASSIGNED=1,
  unassigned issues are now routed by category with logging.
- Log each dispatch decision (issue, agent, category, score, reason) to
  support future audit cycles and accuracy validation.

Closes #476

This operationalizes PR #465's dispatch router into the daily triage loop.
2026-04-25 20:28:33 -04:00
2 changed files with 247 additions and 3 deletions

View File

@@ -129,15 +129,35 @@ Preserved by timmy-orchestrator to prevent loss." 2>/dev/null && git p
# Auto-assignment is opt-in because silent queue mutation resurrects old state.
if [ "$unassigned_count" -gt 0 ]; then
if [ "$AUTO_ASSIGN_UNASSIGNED" = "1" ]; then
log "Assigning $unassigned_count issues to claude..."
log "Routing $unassigned_count issues via dispatch_router..."
DISPATCHER="$SCRIPT_DIR/../tools/dispatch_router.py"
while IFS= read -r line; do
local repo=$(echo "$line" | sed 's/.*REPO=\([^ ]*\).*/\1/')
local num=$(echo "$line" | sed 's/.*NUM=\([^ ]*\).*/\1/')
local title=$(echo "$line" | sed 's/.*TITLE=//')
# Get routing recommendation from dispatch_router
local agent="timmy" category="unknown" score="N/A" reason="unavailable"
if [ -x "$DISPATCHER" ]; then
local result
result=$(python3 "$DISPATCHER" "$title" 2>/dev/null)
if [ $? -eq 0 ]; then
agent=$(echo "$result" | python3 -c "import sys,json; print(json.load(sys.stdin).get('primary',{}).get('agent','timmy'))" 2>/dev/null)
category=$(echo "$result" | python3 -c "import sys,json; print(json.load(sys.stdin).get('category','unknown'))" 2>/dev/null)
local score_raw
score_raw=$(echo "$result" | python3 -c "import sys,json; print(json.load(sys.stdin).get('primary',{}).get('score','N/A'))" 2>/dev/null)
score=$(printf "%.2f" "$score_raw" 2>/dev/null || echo "$score_raw")
reason=$(echo "$result" | python3 -c "import sys,json; print(json.load(sys.stdin).get('primary',{}).get('reason',''))" 2>/dev/null)
[ -z "$agent" ] && agent="timmy"
fi
else
log "WARNING: dispatch_router.py not found or not executable at $DISPATCHER"
fi
# Apply assignment
curl -sf -X PATCH "$GITEA_URL/api/v1/repos/$repo/issues/$num" \
-H "Authorization: token $GITEA_TOKEN" \
-H "Content-Type: application/json" \
-d '{"assignees":["claude"]}' >/dev/null 2>&1 && \
log " Assigned #$num ($repo) to claude"
-d "{\"assignees\":[\"$agent\"]}" >/dev/null 2>&1 && \
log " Routed #$num ($repo) to $agent | category=$category score=$score | $reason"
done < "$state_dir/unassigned.txt"
else
log "Auto-assign disabled: leaving $unassigned_count unassigned issues untouched"

224
tools/dispatch_router.py Normal file
View File

@@ -0,0 +1,224 @@
#!/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()