249 lines
9.2 KiB
Python
249 lines
9.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Small, dependency-free client for the Stackchain Lab Gitea issue loop."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
DEFAULT_BASE = "https://forge.alexanderwhitestone.com/git/api/v1"
|
|
DEFAULT_REPO = "stackchain/stackchain-lab-loop"
|
|
AGENTS = {"timmy", "vincent"}
|
|
STATES = {"ready", "claimed", "review", "blocked"}
|
|
|
|
|
|
class LoopError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class Gitea:
|
|
def __init__(self, base: str, repo: str, token: str):
|
|
self.base = base.rstrip("/")
|
|
self.repo = repo.strip("/")
|
|
self.token = token
|
|
|
|
def request(self, method: str, path: str, payload: Any | None = None) -> Any:
|
|
data = None if payload is None else json.dumps(payload).encode()
|
|
req = urllib.request.Request(
|
|
f"{self.base}{path}",
|
|
data=data,
|
|
method=method,
|
|
headers={
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json",
|
|
"Authorization": f"token {self.token}",
|
|
"User-Agent": "stackchain-lab-loop/0.1",
|
|
},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=20) as response:
|
|
raw = response.read()
|
|
return json.loads(raw) if raw else None
|
|
except urllib.error.HTTPError as exc:
|
|
detail = exc.read().decode(errors="replace")
|
|
raise LoopError(f"Gitea {method} {path} failed: HTTP {exc.code}: {detail}") from exc
|
|
except urllib.error.URLError as exc:
|
|
raise LoopError(f"Gitea {method} {path} failed: {exc.reason}") from exc
|
|
|
|
def labels(self) -> dict[str, int]:
|
|
rows = self.request("GET", f"/repos/{self.repo}/labels?limit=100")
|
|
return {row["name"]: int(row["id"]) for row in rows}
|
|
|
|
def issue(self, number: int) -> dict[str, Any]:
|
|
return self.request("GET", f"/repos/{self.repo}/issues/{number}")
|
|
|
|
def issues(self) -> list[dict[str, Any]]:
|
|
return self.request("GET", f"/repos/{self.repo}/issues?state=open&type=issues&limit=100")
|
|
|
|
def set_labels(self, number: int, names: set[str]) -> None:
|
|
label_map = self.labels()
|
|
missing = sorted(names - label_map.keys())
|
|
if missing:
|
|
raise LoopError(f"Missing repository labels: {', '.join(missing)}")
|
|
self.request(
|
|
"PUT",
|
|
f"/repos/{self.repo}/issues/{number}/labels",
|
|
{"labels": [label_map[name] for name in sorted(names)]},
|
|
)
|
|
|
|
def set_assignee(self, number: int, agent: str) -> None:
|
|
if agent not in AGENTS:
|
|
raise LoopError(f"Unsupported assignee: {agent}")
|
|
self.request("PATCH", f"/repos/{self.repo}/issues/{number}", {"assignee": agent})
|
|
|
|
def comment(self, number: int, body: str) -> None:
|
|
if not body.strip():
|
|
raise LoopError("Refusing to post an empty comment")
|
|
self.request("POST", f"/repos/{self.repo}/issues/{number}/comments", {"body": body.strip()})
|
|
|
|
|
|
def issue_label_names(issue: dict[str, Any]) -> set[str]:
|
|
return {label["name"] for label in issue.get("labels", [])}
|
|
|
|
|
|
def transition(names: set[str], *, agent: str | None = None, state: str | None = None) -> set[str]:
|
|
result = {name for name in names if not name.startswith("agent:") and not name.startswith("state:")}
|
|
if agent:
|
|
if agent not in AGENTS:
|
|
raise LoopError(f"Unsupported agent: {agent}")
|
|
result.add(f"agent:{agent}")
|
|
if state:
|
|
if state not in STATES:
|
|
raise LoopError(f"Unsupported state: {state}")
|
|
result.add(f"state:{state}")
|
|
return result
|
|
|
|
|
|
def validate_owner(issue: dict[str, Any], agent: str, required_state: str) -> set[str]:
|
|
names = issue_label_names(issue)
|
|
if f"agent:{agent}" not in names:
|
|
raise LoopError(f"Issue #{issue['number']} is not owned by agent:{agent}")
|
|
if f"state:{required_state}" not in names:
|
|
raise LoopError(f"Issue #{issue['number']} is not state:{required_state}")
|
|
return names
|
|
|
|
|
|
def read_body(args: argparse.Namespace, default: str = "") -> str:
|
|
if getattr(args, "file", None):
|
|
return Path(args.file).read_text()
|
|
if getattr(args, "body", None):
|
|
return args.body
|
|
return default
|
|
|
|
|
|
def cmd_next(api: Gitea, agent: str, _args: argparse.Namespace) -> int:
|
|
candidates = []
|
|
for issue in api.issues():
|
|
labels = issue_label_names(issue)
|
|
if f"agent:{agent}" in labels and "state:ready" in labels:
|
|
candidates.append(issue)
|
|
candidates.sort(key=lambda row: int(row["number"]))
|
|
if not candidates:
|
|
print(json.dumps({"status": "idle", "agent": agent}))
|
|
return 3
|
|
issue = candidates[0]
|
|
print(json.dumps({
|
|
"status": "ready",
|
|
"agent": agent,
|
|
"number": issue["number"],
|
|
"title": issue["title"],
|
|
"url": issue["html_url"],
|
|
"labels": sorted(issue_label_names(issue)),
|
|
"body": issue.get("body") or "",
|
|
}, indent=2))
|
|
return 0
|
|
|
|
|
|
def cmd_claim(api: Gitea, agent: str, args: argparse.Namespace) -> int:
|
|
issue = api.issue(args.number)
|
|
names = validate_owner(issue, agent, "ready")
|
|
api.set_labels(args.number, transition(names, agent=agent, state="claimed"))
|
|
api.comment(args.number, f"[CLAIM] agent={agent}")
|
|
verified = api.issue(args.number)
|
|
validate_owner(verified, agent, "claimed")
|
|
print(json.dumps({"status": "claimed", "number": args.number, "agent": agent}))
|
|
return 0
|
|
|
|
|
|
def cmd_comment(api: Gitea, agent: str, args: argparse.Namespace) -> int:
|
|
issue = api.issue(args.number)
|
|
validate_owner(issue, agent, "claimed")
|
|
api.comment(args.number, read_body(args))
|
|
print(json.dumps({"status": "commented", "number": args.number, "agent": agent}))
|
|
return 0
|
|
|
|
|
|
def cmd_handoff(api: Gitea, agent: str, args: argparse.Namespace) -> int:
|
|
if args.to == agent:
|
|
raise LoopError("Refusing a handoff to the same agent")
|
|
issue = api.issue(args.number)
|
|
names = validate_owner(issue, agent, "claimed")
|
|
body = read_body(args, f"Continue the bounded task in issue #{args.number}.")
|
|
api.comment(args.number, f"[HANDOFF] from={agent} to={args.to}\n\n{body.strip()}")
|
|
api.set_labels(args.number, transition(names, agent=args.to, state="ready"))
|
|
api.set_assignee(args.number, args.to)
|
|
verified = api.issue(args.number)
|
|
validate_owner(verified, args.to, "ready")
|
|
print(json.dumps({"status": "handed_off", "number": args.number, "from": agent, "to": args.to}))
|
|
return 0
|
|
|
|
|
|
def cmd_finish(api: Gitea, agent: str, args: argparse.Namespace) -> int:
|
|
issue = api.issue(args.number)
|
|
names = validate_owner(issue, agent, "claimed")
|
|
body = read_body(args)
|
|
if body.strip():
|
|
api.comment(args.number, f"[REVIEW] agent={agent}\n\n{body.strip()}")
|
|
api.set_labels(args.number, transition(names, agent=agent, state="review"))
|
|
verified = api.issue(args.number)
|
|
validate_owner(verified, agent, "review")
|
|
print(json.dumps({"status": "review", "number": args.number, "agent": agent}))
|
|
return 0
|
|
|
|
|
|
def cmd_block(api: Gitea, agent: str, args: argparse.Namespace) -> int:
|
|
issue = api.issue(args.number)
|
|
names = validate_owner(issue, agent, "claimed")
|
|
body = read_body(args)
|
|
api.comment(args.number, f"[BLOCKED] agent={agent}\n\n{body.strip()}")
|
|
api.set_labels(args.number, transition(names, agent=agent, state="blocked"))
|
|
print(json.dumps({"status": "blocked", "number": args.number, "agent": agent}))
|
|
return 0
|
|
|
|
|
|
def parser() -> argparse.ArgumentParser:
|
|
p = argparse.ArgumentParser(description=__doc__)
|
|
p.add_argument("--base", default=os.getenv("GITEA_BASE", DEFAULT_BASE))
|
|
p.add_argument("--repo", default=os.getenv("LAB_REPO", DEFAULT_REPO))
|
|
p.add_argument("--agent", default=os.getenv("LAB_AGENT", ""))
|
|
sub = p.add_subparsers(dest="command", required=True)
|
|
sub.add_parser("next")
|
|
for name in ("claim", "comment", "finish", "block"):
|
|
cmd = sub.add_parser(name)
|
|
cmd.add_argument("number", type=int)
|
|
if name != "claim":
|
|
source = cmd.add_mutually_exclusive_group(required=True)
|
|
source.add_argument("--body")
|
|
source.add_argument("--file")
|
|
handoff = sub.add_parser("handoff")
|
|
handoff.add_argument("number", type=int)
|
|
handoff.add_argument("--to", required=True, choices=sorted(AGENTS))
|
|
source = handoff.add_mutually_exclusive_group()
|
|
source.add_argument("--body")
|
|
source.add_argument("--file")
|
|
return p
|
|
|
|
|
|
def main() -> int:
|
|
args = parser().parse_args()
|
|
if args.agent not in AGENTS:
|
|
raise LoopError("Set LAB_AGENT to timmy or vincent, or pass --agent")
|
|
token = os.getenv("GITEA_TOKEN", "")
|
|
if not token:
|
|
raise LoopError("GITEA_TOKEN is required")
|
|
api = Gitea(args.base, args.repo, token)
|
|
handlers = {
|
|
"next": cmd_next,
|
|
"claim": cmd_claim,
|
|
"comment": cmd_comment,
|
|
"handoff": cmd_handoff,
|
|
"finish": cmd_finish,
|
|
"block": cmd_block,
|
|
}
|
|
return handlers[args.command](api, args.agent, args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
raise SystemExit(main())
|
|
except LoopError as exc:
|
|
print(json.dumps({"status": "error", "error": str(exc)}), file=sys.stderr)
|
|
raise SystemExit(2)
|