forked from Rockachopa/Timmy-time-dashboard
245 lines
7.4 KiB
Python
245 lines
7.4 KiB
Python
"""Vassal Protocol — agent dispatch.
|
|
|
|
Translates triage decisions into concrete Gitea actions:
|
|
- Add ``claude-ready`` or ``kimi-ready`` label to an issue
|
|
- Post a dispatch comment recording the routing rationale
|
|
- Record the dispatch in the in-memory registry so the orchestration loop
|
|
can track what was sent and when
|
|
|
|
The dispatch registry is intentionally in-memory (ephemeral). Durable
|
|
tracking is out of scope for this module — that belongs in the task queue
|
|
or a future orchestration DB.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import dataclass, field
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from timmy.vassal.backlog import AgentTarget, TriagedIssue
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Label names used by the dispatch system
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_LABEL_MAP: dict[AgentTarget, str] = {
|
|
AgentTarget.CLAUDE: "claude-ready",
|
|
AgentTarget.KIMI: "kimi-ready",
|
|
AgentTarget.TIMMY: "timmy-ready",
|
|
}
|
|
|
|
_LABEL_COLORS: dict[str, str] = {
|
|
"claude-ready": "#8b6f47", # warm brown
|
|
"kimi-ready": "#006b75", # dark teal
|
|
"timmy-ready": "#0075ca", # blue
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dispatch registry
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class DispatchRecord:
|
|
"""A record of one issue being dispatched to an agent."""
|
|
|
|
issue_number: int
|
|
issue_title: str
|
|
agent: AgentTarget
|
|
rationale: str
|
|
dispatched_at: str = field(default_factory=lambda: datetime.now(UTC).isoformat())
|
|
label_applied: bool = False
|
|
comment_posted: bool = False
|
|
|
|
|
|
# Module-level registry: issue_number → DispatchRecord
|
|
_registry: dict[int, DispatchRecord] = {}
|
|
|
|
|
|
def get_dispatch_registry() -> dict[int, DispatchRecord]:
|
|
"""Return a copy of the current dispatch registry."""
|
|
return dict(_registry)
|
|
|
|
|
|
def clear_dispatch_registry() -> None:
|
|
"""Clear the dispatch registry (mainly for tests)."""
|
|
_registry.clear()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Gitea helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def _get_or_create_label(
|
|
client: Any,
|
|
base_url: str,
|
|
headers: dict,
|
|
repo: str,
|
|
label_name: str,
|
|
) -> int | None:
|
|
"""Return the Gitea label ID, creating it if necessary."""
|
|
labels_url = f"{base_url}/repos/{repo}/labels"
|
|
try:
|
|
resp = await client.get(labels_url, headers=headers)
|
|
if resp.status_code == 200:
|
|
for lbl in resp.json():
|
|
if lbl.get("name") == label_name:
|
|
return lbl["id"]
|
|
except Exception as exc:
|
|
logger.warning("_get_or_create_label: list failed — %s", exc)
|
|
return None
|
|
|
|
color = _LABEL_COLORS.get(label_name, "#cccccc")
|
|
try:
|
|
resp = await client.post(
|
|
labels_url,
|
|
headers={**headers, "Content-Type": "application/json"},
|
|
json={"name": label_name, "color": color},
|
|
)
|
|
if resp.status_code in (200, 201):
|
|
return resp.json().get("id")
|
|
except Exception as exc:
|
|
logger.warning("_get_or_create_label: create failed — %s", exc)
|
|
|
|
return None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dispatch action helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def _apply_label_to_issue(
|
|
client: Any,
|
|
base_url: str,
|
|
headers: dict,
|
|
repo: str,
|
|
issue_number: int,
|
|
label_name: str,
|
|
) -> bool:
|
|
"""Get-or-create the label then apply it to the issue. Returns True on success."""
|
|
label_id = await _get_or_create_label(client, base_url, headers, repo, label_name)
|
|
if label_id is None:
|
|
return False
|
|
resp = await client.post(
|
|
f"{base_url}/repos/{repo}/issues/{issue_number}/labels",
|
|
headers=headers,
|
|
json={"labels": [label_id]},
|
|
)
|
|
return resp.status_code in (200, 201)
|
|
|
|
|
|
async def _post_dispatch_comment(
|
|
client: Any,
|
|
base_url: str,
|
|
headers: dict,
|
|
repo: str,
|
|
issue: TriagedIssue,
|
|
label_name: str,
|
|
) -> bool:
|
|
"""Post the vassal routing comment. Returns True on success."""
|
|
agent_name = issue.agent_target.value.capitalize()
|
|
comment_body = (
|
|
f"🤖 **Vassal dispatch** → routed to **{agent_name}**\n\n"
|
|
f"Priority score: {issue.priority_score} \n"
|
|
f"Rationale: {issue.rationale} \n"
|
|
f"Label: `{label_name}`"
|
|
)
|
|
resp = await client.post(
|
|
f"{base_url}/repos/{repo}/issues/{issue.number}/comments",
|
|
headers=headers,
|
|
json={"body": comment_body},
|
|
)
|
|
return resp.status_code in (200, 201)
|
|
|
|
|
|
async def _perform_gitea_dispatch(
|
|
issue: TriagedIssue,
|
|
record: DispatchRecord,
|
|
) -> None:
|
|
"""Apply label and post comment via Gitea. Mutates *record* in-place."""
|
|
try:
|
|
import httpx
|
|
|
|
from config import settings
|
|
except ImportError as exc:
|
|
logger.warning("dispatch_issue: missing dependency — %s", exc)
|
|
return
|
|
|
|
if not settings.gitea_enabled or not settings.gitea_token:
|
|
logger.info("dispatch_issue: Gitea disabled — skipping label/comment")
|
|
return
|
|
|
|
base_url = f"{settings.gitea_url}/api/v1"
|
|
repo = settings.gitea_repo
|
|
headers = {
|
|
"Authorization": f"token {settings.gitea_token}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
label_name = _LABEL_MAP[issue.agent_target]
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
record.label_applied = await _apply_label_to_issue(
|
|
client, base_url, headers, repo, issue.number, label_name
|
|
)
|
|
record.comment_posted = await _post_dispatch_comment(
|
|
client, base_url, headers, repo, issue, label_name
|
|
)
|
|
except Exception as exc:
|
|
logger.warning("dispatch_issue: Gitea action failed — %s", exc)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dispatch action
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def dispatch_issue(issue: TriagedIssue) -> DispatchRecord:
|
|
"""Apply dispatch label and post a routing comment on the Gitea issue.
|
|
|
|
Gracefully degrades: if Gitea is unavailable the record is still
|
|
created and returned (with label_applied=False, comment_posted=False).
|
|
|
|
Args:
|
|
issue: A TriagedIssue with a routing decision.
|
|
|
|
Returns:
|
|
DispatchRecord summarising what was done.
|
|
"""
|
|
record = DispatchRecord(
|
|
issue_number=issue.number,
|
|
issue_title=issue.title,
|
|
agent=issue.agent_target,
|
|
rationale=issue.rationale,
|
|
)
|
|
|
|
if issue.agent_target == AgentTarget.TIMMY:
|
|
# Self-dispatch: no label needed — Timmy will handle directly.
|
|
logger.info(
|
|
"dispatch_issue: #%d '%s' → Timmy (self, no label)",
|
|
issue.number,
|
|
issue.title[:50],
|
|
)
|
|
_registry[issue.number] = record
|
|
return record
|
|
|
|
await _perform_gitea_dispatch(issue, record)
|
|
|
|
_registry[issue.number] = record
|
|
logger.info(
|
|
"dispatch_issue: #%d '%s' → %s (label=%s comment=%s)",
|
|
issue.number,
|
|
issue.title[:50],
|
|
issue.agent_target,
|
|
record.label_applied,
|
|
record.comment_posted,
|
|
)
|
|
return record
|