feat: add Gitea task delegation scaffold (#550)

scripts/gitea_task_delegator.py

@@ -0,0 +1,253 @@
+#!/usr/bin/env python3
+"""Cross-agent task delegator for Gitea issues.
+
+Refs: timmy-home #550
+
+Phase-3 coordination slice:
+- inspect open Gitea issues
+- route clear work items to wizard houses
+- assign through Gitea when explicitly applied
+- stay conservative on ambiguous issues
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+from urllib import request
+from typing import Any
+
+DEFAULT_BASE_URL = "https://forge.alexanderwhitestone.com/api/v1"
+DEFAULT_OWNER = "Timmy_Foundation"
+DEFAULT_REPO = "timmy-home"
+DEFAULT_TOKEN_FILE = Path.home() / ".config" / "gitea" / "token"
+
+ROUTING_RULES = {
+    "timmy": {
+        "route": "critical",
+        "label_terms": {"critical", "security", "policy", "sovereign", "urgent", "p0"},
+        "title_terms": [
+            "critical", "security", "sovereign", "policy", "final decision", "review gate", "approval",
+        ],
+        "body_terms": [
+            "critical", "security", "sovereign", "final decision", "manual review", "approval",
+        ],
+    },
+    "ezra": {
+        "route": "documentation",
+        "label_terms": {"documentation", "docs", "analysis", "research", "audit", "genome"},
+        "title_terms": [
+            "documentation", "docs", "analysis", "research", "audit", "genome", "architecture", "readme",
+        ],
+        "body_terms": [
+            "documentation", "analysis", "research", "audit", "architecture", "writeup", "report",
+        ],
+    },
+    "bezalel": {
+        "route": "implementation",
+        "label_terms": {"bug", "feature", "testing", "tests", "ci", "build", "fix"},
+        "title_terms": [
+            "fix", "build", "test", "tests", "ci", "feature", "implementation", "deploy", "pipeline",
+        ],
+        "body_terms": [
+            "implementation", "testing", "build", "ci", "fix", "feature", "deploy", "proof",
+        ],
+    },
+    "allegro": {
+        "route": "routing",
+        "label_terms": {"ops", "routing", "dispatch", "coordination", "connectivity", "orchestration"},
+        "title_terms": [
+            "routing", "dispatch", "coordination", "connectivity", "orchestration", "queue", "tempo", "handoff",
+        ],
+        "body_terms": [
+            "routing", "dispatch", "coordination", "connectivity", "orchestration", "agent-to-agent", "cross-fleet",
+        ],
+    },
+}
+
+
+class GiteaClient:
+    def __init__(self, token: str, owner: str = DEFAULT_OWNER, repo: str = DEFAULT_REPO, base_url: str = DEFAULT_BASE_URL):
+        self.token = token
+        self.owner = owner
+        self.repo = repo
+        self.base_url = base_url.rstrip("/")
+
+    def _request(self, path: str, *, method: str = "GET", data: dict[str, Any] | None = None):
+        payload = None if data is None else json.dumps(data).encode()
+        headers = {"Authorization": f"token {self.token}"}
+        if payload is not None:
+            headers["Content-Type"] = "application/json"
+        req = request.Request(f"{self.base_url}{path}", data=payload, headers=headers, method=method)
+        with request.urlopen(req, timeout=30) as resp:
+            return json.loads(resp.read().decode())
+
+    def list_open_issues(self, limit: int = 100):
+        issues = self._request(f"/repos/{self.owner}/{self.repo}/issues?state=open&limit={limit}")
+        return [issue for issue in issues if not issue.get("pull_request")]
+
+    def assign_issue(self, issue_number: int, assignee: str):
+        return self._request(
+            f"/repos/{self.owner}/{self.repo}/issues/{issue_number}",
+            method="PATCH",
+            data={"assignees": [assignee]},
+        )
+
+
+def _labels(issue: dict[str, Any]) -> set[str]:
+    return {str(label.get("name", "")).strip().lower() for label in (issue.get("labels") or []) if label.get("name")}
+
+
+def _score_rule(text_title: str, text_body: str, labels: set[str], rule: dict[str, Any]):
+    score = 0
+    matched_terms: list[str] = []
+
+    for term in sorted(rule["label_terms"]):
+        if term in labels:
+            score += 3
+            matched_terms.append(term)
+
+    for term in rule["title_terms"]:
+        if term in text_title:
+            score += 2
+            matched_terms.append(term)
+
+    for term in rule["body_terms"]:
+        if term in text_body:
+            score += 1
+            matched_terms.append(term)
+
+    deduped = []
+    seen = set()
+    for term in matched_terms:
+        if term in seen:
+            continue
+        seen.add(term)
+        deduped.append(term)
+
+    return score, deduped
+
+
+def classify_issue(issue: dict[str, Any], minimum_confidence: int = 3):
+    title = str(issue.get("title") or "").lower()
+    body = str(issue.get("body") or "").lower()
+    labels = _labels(issue)
+
+    scored = []
+    for assignee, rule in ROUTING_RULES.items():
+        score, matched_terms = _score_rule(title, body, labels, rule)
+        if score > 0:
+            scored.append((score, assignee, rule["route"], matched_terms))
+
+    if not scored:
+        return None
+
+    scored.sort(key=lambda item: (-item[0], item[1]))
+    best_score, best_assignee, route, matched_terms = scored[0]
+    if best_score < minimum_confidence:
+        return None
+    if len(scored) > 1 and scored[1][0] == best_score:
+        return None
+
+    return {
+        "assignee": best_assignee,
+        "route": route,
+        "confidence": best_score,
+        "matched_terms": matched_terms,
+    }
+
+
+def build_assignment_plan(issues: list[dict[str, Any]], minimum_confidence: int = 3):
+    assignments = []
+    skipped = []
+
+    for issue in issues:
+        issue_number = issue.get("number")
+        if issue.get("pull_request"):
+            skipped.append({"issue_number": issue_number, "reason": "pull_request"})
+            continue
+
+        assignees = issue.get("assignees") or []
+        if assignees:
+            skipped.append({"issue_number": issue_number, "reason": "already_assigned"})
+            continue
+
+        recommendation = classify_issue(issue, minimum_confidence=minimum_confidence)
+        if recommendation is None:
+            skipped.append({"issue_number": issue_number, "reason": "no_confident_route"})
+            continue
+
+        assignments.append({
+            "issue_number": issue_number,
+            **recommendation,
+        })
+
+    assignments.sort(key=lambda item: (-item["confidence"], item["issue_number"]))
+    return {"assignments": assignments, "skipped": skipped}
+
+
+def apply_assignment_plan(plan: dict[str, Any], client: GiteaClient, apply: bool = False):
+    results = []
+    for item in plan.get("assignments", []):
+        if apply:
+            client.assign_issue(item["issue_number"], item["assignee"])
+            results.append({
+                "action": "assigned",
+                "issue_number": item["issue_number"],
+                "assignee": item["assignee"],
+            })
+        else:
+            results.append({
+                "action": "would_assign",
+                "issue_number": item["issue_number"],
+                "assignee": item["assignee"],
+            })
+    return results
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(description="Route open Gitea issues to wizard houses and optionally assign them.")
+    parser.add_argument("--owner", default=DEFAULT_OWNER)
+    parser.add_argument("--repo", default=DEFAULT_REPO)
+    parser.add_argument("--base-url", default=DEFAULT_BASE_URL)
+    parser.add_argument("--token-file", type=Path, default=DEFAULT_TOKEN_FILE)
+    parser.add_argument("--limit", type=int, default=100)
+    parser.add_argument("--minimum-confidence", type=int, default=3)
+    parser.add_argument("--apply", action="store_true", help="Apply assignments to Gitea instead of reporting them.")
+    parser.add_argument("--json", action="store_true", help="Emit machine-readable JSON output.")
+    return parser.parse_args()
+
+
+def main():
+    args = parse_args()
+    if not args.token_file.exists():
+        raise SystemExit(f"Token file not found: {args.token_file}")
+
+    token = args.token_file.read_text().strip()
+    client = GiteaClient(token=token, owner=args.owner, repo=args.repo, base_url=args.base_url)
+    issues = client.list_open_issues(limit=args.limit)
+    plan = build_assignment_plan(issues, minimum_confidence=args.minimum_confidence)
+    results = apply_assignment_plan(plan, client, apply=args.apply)
+    payload = {
+        "assignments": plan["assignments"],
+        "skipped": plan["skipped"],
+        "results": results,
+    }
+
+    if args.json:
+        print(json.dumps(payload, indent=2))
+        return
+
+    print("--- Gitea Task Delegator ---")
+    print(f"Assignments: {len(plan['assignments'])}")
+    for item in plan["assignments"]:
+        print(f"- #{item['issue_number']} -> {item['assignee']} ({item['route']}, confidence={item['confidence']})")
+    print(f"Skipped: {len(plan['skipped'])}")
+    for item in plan["skipped"][:20]:
+        print(f"- #{item['issue_number']}: {item['reason']}")
+
+
+if __name__ == "__main__":
+    main()

tests/test_gitea_task_delegator.py

@@ -0,0 +1,124 @@
+from scripts.gitea_task_delegator import (
+    apply_assignment_plan,
+    build_assignment_plan,
+    classify_issue,
+)
+
+
+class FakeGiteaClient:
+    def __init__(self):
+        self.assignments = []
+
+    def assign_issue(self, issue_number, assignee):
+        payload = {"number": issue_number, "assignee": assignee}
+        self.assignments.append(payload)
+        return payload
+
+
+def make_issue(number, title, body="", labels=None, assignees=None, pull_request=None):
+    return {
+        "number": number,
+        "title": title,
+        "body": body,
+        "labels": [{"name": name} for name in (labels or [])],
+        "assignees": list(assignees or []),
+        "pull_request": pull_request,
+    }
+
+
+def test_classify_issue_routes_documentation_to_ezra():
+    issue = make_issue(
+        11,
+        "Codebase genome docs pass",
+        body="Need architecture analysis and documentation updates.",
+        labels=["documentation"],
+    )
+
+    recommendation = classify_issue(issue)
+
+    assert recommendation["assignee"] == "ezra"
+    assert recommendation["route"] == "documentation"
+    assert "documentation" in recommendation["matched_terms"]
+
+
+def test_classify_issue_routes_build_and_test_work_to_bezalel():
+    issue = make_issue(
+        12,
+        "Fix failing CI tests",
+        body="Implementation and testing work needed for the build lane.",
+        labels=["bug"],
+    )
+
+    recommendation = classify_issue(issue)
+
+    assert recommendation["assignee"] == "bezalel"
+    assert recommendation["route"] == "implementation"
+
+
+def test_classify_issue_routes_connectivity_work_to_allegro():
+    issue = make_issue(
+        13,
+        "Dispatch routing broken between hosts",
+        body="Need connectivity and coordination fixes for cross-host dispatch.",
+        labels=["ops"],
+    )
+
+    recommendation = classify_issue(issue)
+
+    assert recommendation["assignee"] == "allegro"
+    assert recommendation["route"] == "routing"
+
+
+def test_classify_issue_routes_critical_policy_work_to_timmy():
+    issue = make_issue(
+        14,
+        "Critical sovereign policy decision",
+        body="Security-sensitive final decision required.",
+        labels=["critical"],
+    )
+
+    recommendation = classify_issue(issue)
+
+    assert recommendation["assignee"] == "timmy"
+    assert recommendation["route"] == "critical"
+
+
+def test_build_assignment_plan_skips_assigned_prs_and_ambiguous_issues():
+    issues = [
+        make_issue(1, "Write architecture docs", labels=["documentation"]),
+        make_issue(2, "Already owned", labels=["bug"], assignees=[{"login": "bezalel"}]),
+        make_issue(3, "misc cleanup"),
+        make_issue(4, "Docs PR", labels=["documentation"], pull_request={"url": "https://example/pr/4"}),
+    ]
+
+    plan = build_assignment_plan(issues)
+
+    assert [item["issue_number"] for item in plan["assignments"]] == [1]
+    assert plan["assignments"][0]["assignee"] == "ezra"
+    assert plan["skipped"] == [
+        {"issue_number": 2, "reason": "already_assigned"},
+        {"issue_number": 3, "reason": "no_confident_route"},
+        {"issue_number": 4, "reason": "pull_request"},
+    ]
+
+
+def test_apply_assignment_plan_supports_dry_run_and_apply_modes():
+    plan = {
+        "assignments": [
+            {"issue_number": 21, "assignee": "ezra", "route": "documentation", "confidence": 7},
+            {"issue_number": 22, "assignee": "allegro", "route": "routing", "confidence": 6},
+        ],
+        "skipped": [],
+    }
+    client = FakeGiteaClient()
+
+    dry_run = apply_assignment_plan(plan, client, apply=False)
+    assert [item["action"] for item in dry_run] == ["would_assign", "would_assign"]
+    assert client.assignments == []
+
+    applied = apply_assignment_plan(plan, client, apply=True)
+    assert [item["action"] for item in applied] == ["assigned", "assigned"]
+    assert client.assignments == [
+        {"number": 21, "assignee": "ezra"},
+        {"number": 22, "assignee": "allegro"},
+    ]