stackchain-lab-loop/scripts/lab_loop.py

#!/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)