#!/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( "PATCH", f"/repos/{self.repo}/issues/{number}", {"labels": [label_map[name] for name in sorted(names)]}, ) 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")) 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)